已合并
Merge branch '0506_enter' into 0506_dev #210
Merge branch '0506_enter' into 0506_dev #210
已合并
zhangdanyang创建于 5月6日
51 个文件变更+6555-284
@@ -16,16 +16,16 @@ class Settings(BaseSettings):
16 # --------------------------16 # --------------------------
17 # 【基础配置】17 # 【基础配置】
18 # --------------------------18 # --------------------------
19- DB_TYPE: Literal["mysql", "sqlite", "gaussdb", "opengauss"] = Field(default="sqlite", env="DB_TYPE")19+ RUNTIME_DB_TYPE: Literal["mysql", "sqlite", "gaussdb", "opengauss"] = Field(default="sqlite", env="RUNTIME_DB_TYPE")
20 20 
21 # --------------------------21 # --------------------------
22- # 【MySQL 配置】(静态可选,DB_TYPE=mysql 时必选)22+ # 【MySQL RUNTIME_DB_TYPE=mysql 时必选)
23 # --------------------------23 # --------------------------
24- DB_HOST: Optional[str] = Field(default=None, env="DB_HOST")24+ RUNTIME_DB_HOST: Optional[str] = Field(default=None, env="RUNTIME_DB_HOST")
25- DB_PORT: Optional[int] = Field(default=None, env="DB_PORT")25+ RUNTIME_DB_PORT: Optional[int] = Field(default=None, env="RUNTIME_DB_PORT")
26- DB_USER: Optional[str] = Field(default=None, env="DB_USER")26+ RUNTIME_DB_USER: Optional[str] = Field(default=None, env="RUNTIME_DB_USER")
27- DB_PASSWORD: Optional[str] = Field(default=None, env="DB_PASSWORD")27+ RUNTIME_DB_PASSWORD: Optional[str] = Field(default=None, env="RUNTIME_DB_PASSWORD")
28- DB_NAME: Optional[str] = Field(default=None, env="DB_NAME")28+ RUNTIME_DB_NAME: Optional[str] = Field(default=None, env="RUNTIME_DB_NAME")
29 29 
30 # --------------------------30 # --------------------------
31 # 【服务配置】31 # 【服务配置】
@@ -53,23 +53,26 @@ class Settings(BaseSettings):
53 # ========================53 # ========================
54 @model_validator(mode="after")54 @model_validator(mode="after")
55 def check_mysql_required(self) -> "Settings":55 def check_mysql_required(self) -> "Settings":
56- if self.DB_TYPE in {"mysql", "gaussdb", "opengauss"}:56+ if self.RUNTIME_DB_TYPE in {"mysql", "gaussdb", "opengauss"}:
57 missing = []57 missing = []
58- if not self.DB_HOST:58+ if not self.RUNTIME_DB_HOST:
59- missing.append("DB_HOST")59+ missing.append("RUNTIME_DB_HOST")
60- if not self.DB_PORT:60+ if not self.RUNTIME_DB_PORT:
61- missing.append("DB_PORT")61+ missing.append("RUNTIME_DB_PORT")
62- if not self.DB_USER:62+ if not self.RUNTIME_DB_USER:
63- missing.append("DB_USER")63+ missing.append("RUNTIME_DB_USER")
64- if not self.DB_PASSWORD:64+ if not self.RUNTIME_DB_PASSWORD:
65- missing.append("DB_PASSWORD")65+ missing.append("RUNTIME_DB_PASSWORD")
66- if not self.DB_NAME:66+ if not self.RUNTIME_DB_NAME:
67- missing.append("DB_NAME")67+ missing.append("RUNTIME_DB_NAME")
68 68 
69 if missing:69 if missing:
70- raise ValueError(70+ msg = (
71- f"When DB_TYPE is mysql/gaussdb/opengauss, the following fields are required: {', '.join(missing)}"71+ "When RUNTIME_DB_TYPE is mysql/gaussdb/opengauss, "
72+ "the following fields are required: "
73+ f"{', '.join(missing)}"
72 )74 )
75+ raise ValueError(msg)
73 return self76 return self
74 77 
75 @model_validator(mode="after")78 @model_validator(mode="after")
@@ -130,7 +133,8 @@ class Settings(BaseSettings):
130 model_config = SettingsConfigDict(133 model_config = SettingsConfigDict(
131 env_file=os.path.join(PROJECT_ROOT, "server/.env"),134 env_file=os.path.join(PROJECT_ROOT, "server/.env"),
132 env_file_encoding="utf-8",135 env_file_encoding="utf-8",
133- case_sensitive=True136+ case_sensitive=True,
137+ extra="ignore"
134 )138 )
135 139 
136# 初始化配置140# 初始化配置
Mfoundation/openjiuwen_runtime/foundation/db/sqlalchemy_handler.py+79-2文件内容审核中,请稍后刷新重试
@@ -18,10 +18,10 @@ dependencies = [
18 "greenlet>=3.0.0",18 "greenlet>=3.0.0",
19 "pymysql==1.1.1",19 "pymysql==1.1.1",
20 "cryptography>=42.0.0",20 "cryptography>=42.0.0",
21- "pydantic==2.11.7",21+ "pydantic>=2.11.7",
22- "pydantic-settings==2.5.2",22+ "pydantic-settings>=2.5.2",
23 "aiomysql==0.2.0",23 "aiomysql==0.2.0",
24- "aiosqlite==0.21.0",24+ "aiosqlite>=0.21.0",
25 "redis==7.1.0",25 "redis==7.1.0",
26 # lint26 # lint
27 "ruff==0.9.10",27 "ruff==0.9.10",
@@ -4,7 +4,6 @@
4"""OpenJiuwen Runtime Management SDK"""4"""OpenJiuwen Runtime Management SDK"""
5 5 
6from openjiuwen_runtime.foundation.db.handler import DBHandler6from openjiuwen_runtime.foundation.db.handler import DBHandler
7-from openjiuwen_runtime.foundation.db.gaussdb_handler import GaussDBHandler
8from openjiuwen_runtime.foundation.db.mysql_handler import MySQLHandler7from openjiuwen_runtime.foundation.db.mysql_handler import MySQLHandler
9from openjiuwen_runtime.foundation.db.sqlite_handler import SQLiteHandler8from openjiuwen_runtime.foundation.db.sqlite_handler import SQLiteHandler
10 9 
@@ -12,6 +11,7 @@ from .manager import DeploymentManager
12from .models.deployment_params import (11from .models.deployment_params import (
13 DeployAgentParams,12 DeployAgentParams,
14 DeployPluginParams,13 DeployPluginParams,
14+ DeployImageParams,
15 ListDeploymentsParams,15 ListDeploymentsParams,
16)16)
17from .models.enums import DeployMode, DeploymentType, DeploymentStatus17from .models.enums import DeployMode, DeploymentType, DeploymentStatus
@@ -49,6 +49,7 @@ __all__ = [
49 # Deployment params49 # Deployment params
50 "DeployAgentParams",50 "DeployAgentParams",
51 "DeployPluginParams",51 "DeployPluginParams",
52+ "DeployImageParams",
52 "ListDeploymentsParams",53 "ListDeploymentsParams",
53 # Enums54 # Enums
54 "DeploymentType",55 "DeploymentType",
@@ -61,7 +62,6 @@ __all__ = [
61 "DBHandler",62 "DBHandler",
62 "SQLiteHandler",63 "SQLiteHandler",
63 "MySQLHandler",64 "MySQLHandler",
64- "GaussDBHandler",
65 # Base65 # Base
66 "CommonParams",66 "CommonParams",
67 "DeployContext",67 "DeployContext",
@@ -32,14 +32,26 @@ from .docker import (
32 DockerDeployer,32 DockerDeployer,
33 DockerStrategy,33 DockerStrategy,
34)34)
35-from .k8s import (35+ 
36- K8sParams,36+K8S_IMPORT_ERROR = None
37- K8sInfo,37+ 
38- K8sCreate,38+try:
39- K8S_TABLE_DEF,39+ from .k8s import (
40- K8sDeployer,40+ K8sParams,
41- K8sStrategy,41+ K8sInfo,
42-)42+ K8sCreate,
43+ K8S_TABLE_DEF,
44+ K8sDeployer,
45+ K8sStrategy,
46+ )
47+except Exception as exc: # pragma: no cover - depends on local runtime availability
48+ K8S_IMPORT_ERROR = exc
49+ K8sParams = None
50+ K8sInfo = None
51+ K8sCreate = None
52+ K8S_TABLE_DEF = None
53+ K8sDeployer = None
54+ K8sStrategy = None
43 55 
44__all__ = [56__all__ = [
45 # Base57 # Base
@@ -63,6 +75,7 @@ __all__ = [
63 "DockerDeployer",75 "DockerDeployer",
64 "DockerStrategy",76 "DockerStrategy",
65 # K8s77 # K8s
78+ "K8S_IMPORT_ERROR",
66 "K8sParams",79 "K8sParams",
67 "K8sInfo",80 "K8sInfo",
68 "K8sCreate",81 "K8sCreate",
@@ -1,243 +1,770 @@
1-# coding: utf-81+"""K8s deployer implemented with the Kubernetes Python client."""
2-# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3- 
4-"""K8s 部署器"""
5 2 
6import asyncio3import asyncio
4+import errno
5+import logging
7import os6import os
8-from typing import Optional7+from pathlib import Path
8+from dataclasses import dataclass
9+from typing import Optional, Any
9 10 
10-from openjiuwen_runtime.foundation.log import get_logger11+from kubernetes import client
12+from kubernetes import config as k8s_config
13+from kubernetes.client.rest import ApiException
14+from kubernetes.config.config_exception import ConfigException
11 15 
12from ..base.deployer import Deployer16from ..base.deployer import Deployer
13from ..base.models import DeployContext, DeployResult17from ..base.models import DeployContext, DeployResult
14from .models import K8sParams18from .models import K8sParams
15from ...models.enums import DeploymentStatus19from ...models.enums import DeploymentStatus
16 20 
17-logger = get_logger(__name__)21+logger = logging.getLogger(__name__)
22+ 
23+ 
24+@dataclass
25+class RuntimeSettings:
26+ """K8s 部署的运行时配置"""
27+ namespace: str
28+ deployment_name: str
29+ replicas: int
30+ container_port: int
31+ service_port: int
32+ service_type: str
33+ node_port: Optional[int]
34+ node_selector: dict[str, Any]
35+ image: str
36+ image_pull_policy: str
37+ userdata: Optional[str]
18 38 
19 39 
20class K8sDeployer(Deployer[K8sParams]):40class K8sDeployer(Deployer[K8sParams]):
21- """Kubernetes 部署器"""41+ """Kubernetes deployer backed by the official Python client."""
42+ 
43+ _WINDOWS_SOCKET_ERROR_MESSAGES = {
44+ 10053: "software caused connection abort",
45+ 10054: "connection reset by peer",
46+ 10060: "connection timed out",
47+ 10061: "connection refused",
48+ }
22 49 
23 def __init__(50 def __init__(
24- self,51+ self,
25- default_host: str = "localhost",52+ default_host: str = "localhost",
26- kubeconfig: Optional[str] = None,53+ kubeconfig: Optional[str] = None,
27- namespace: str = "default",54+ namespace: str = "default",
55+ rollout_timeout: int = 300,
56+ rollout_poll_interval: float = 2.0,
28 ):57 ):
29 self.default_host = default_host58 self.default_host = default_host
30- self.kubeconfig = kubeconfig59+ self.kubeconfig = kubeconfig or os.getenv("KUBECONFIG")
31 self.namespace = namespace60 self.namespace = namespace
61+ self.rollout_timeout = rollout_timeout
62+ self.rollout_poll_interval = rollout_poll_interval
32 self._deployments: dict[str, str] = {}63 self._deployments: dict[str, str] = {}
33- logger.debug("K8sDeployer initialized: namespace=%s", namespace)64+ self._config_loaded = False
34- 65+ logger.info(
35- def _get_kubectl_env(self) -> dict:66+ "K8sClientDeployer initialized: namespace=%s, kubeconfig_configured=%s",
36- env = os.environ.copy()67+ namespace,
37- if self.kubeconfig:68+ bool(self.kubeconfig),
38- env["KUBECONFIG"] = self.kubeconfig
39- return env
40- 
41- async def _run_kubectl_command(self, *args: str) -> tuple[bool, str]:
42- cmd = ["kubectl"]
43- cmd.extend(args)
44- 
45- logger.debug("Running kubectl command: %s", " ".join(args))
46- process = await asyncio.create_subprocess_exec(
47- *cmd,
48- stdout=asyncio.subprocess.PIPE,
49- stderr=asyncio.subprocess.PIPE,
50- env=self._get_kubectl_env(),
51 )69 )
52- stdout, stderr = await process.communicate()
53 70 
54- if process.returncode == 0:71+ @staticmethod
55- return True, stdout.decode().strip()72+ def _default_resource_name(deployment_id: str) -> str:
56- return False, stderr.decode().strip()73+ return f"agent-{deployment_id[:6]}"
74+ 
75+ @staticmethod
76+ def _normalize_mapping(value: Optional[dict[str, Any]]) -> dict[str, Any]:
77+ if isinstance(value, dict):
78+ return value
79+ return {}
80+ 
81+ @staticmethod
82+ def _format_api_exception(exc: ApiException) -> str:
83+ body = getattr(exc, "body", None)
84+ if body:
85+ return f"{exc.status} {exc.reason}: {body}"
86+ return f"{exc.status} {exc.reason}"
87+ 
88+ @classmethod
89+ def _format_os_error(cls, exc: OSError) -> str:
90+ error_number = getattr(exc, "errno", None)
91+ winerror = getattr(exc, "winerror", None)
92+ message = (
93+ cls._WINDOWS_SOCKET_ERROR_MESSAGES.get(winerror)
94+ or cls._WINDOWS_SOCKET_ERROR_MESSAGES.get(error_number)
95+ )
96+ if not message:
97+ code = winerror if winerror is not None else error_number
98+ if code is not None:
99+ message = errno.errorcode.get(code, exc.__class__.__name__)
100+ else:
101+ message = exc.__class__.__name__
102+ 
103+ details = []
104+ if error_number is not None:
105+ details.append(f"errno={error_number}")
106+ if winerror is not None and winerror != error_number:
107+ details.append(f"winerror={winerror}")
108+ if details:
109+ return f"{message} ({', '.join(details)})"
110+ return message
111+ 
112+ @classmethod
113+ def _format_exception_message(cls, exc: BaseException) -> str:
114+ if isinstance(exc, ApiException):
115+ return cls._format_api_exception(exc)
116+ if isinstance(exc, OSError):
117+ return cls._format_os_error(exc)
118+ 
119+ parts: list[str] = []
120+ for arg in getattr(exc, "args", ()):
121+ if isinstance(arg, BaseException):
122+ formatted = cls._format_exception_message(arg)
123+ else:
124+ formatted = str(arg).strip()
125+ if formatted:
126+ parts.append(formatted.rstrip("."))
127+ 
128+ if parts:
129+ return ": ".join(parts)
130+ 
131+ message = str(exc).strip()
132+ return message or exc.__class__.__name__
133+ 
134+ @staticmethod
135+ def _is_not_found(exc: ApiException) -> bool:
136+ return getattr(exc, "status", None) == 404
137+ 
138+ @staticmethod
139+ def _is_conflict(exc: ApiException) -> bool:
140+ return getattr(exc, "status", None) == 409
141+ 
142+ def _get_kubeconfig_host(self) -> Optional[tuple]:
143+ """从 kubeconfig 的 cluster server URL 中提取主机 IP。"""
144+ if not self.kubeconfig:
145+ return None
146+ try:
147+ config_path = Path(self.kubeconfig)
148+ if not config_path.exists():
149+ return None
150+ import yaml
151+ from urllib.parse import urlparse
152+ with open(config_path, "r", encoding="utf-8") as f:
153+ kube_conf = yaml.safe_load(f)
154+ clusters = kube_conf.get("clusters", [])
155+ if not clusters:
156+ return None
157+ server_url = clusters[0].get("cluster", {}).get("server", "")
158+ parsed = urlparse(server_url)
159+ return parsed.hostname, parsed.port
160+ except Exception as exc:
161+ logger.warning("Failed to parse kubeconfig host: %s", exc)
162+ return None
163+ 
164+ async def _call_api(self, func, *args, **kwargs):
165+ return await asyncio.to_thread(func, *args, **kwargs)
166+ 
167+ async def _ensure_client_config(self) -> None:
168+ if self._config_loaded:
169+ return
170+ 
171+ def _load() -> None:
172+ if self.kubeconfig:
173+ k8s_config.load_kube_config(config_file=self.kubeconfig)
174+ return
175+ try:
176+ k8s_config.load_incluster_config()
177+ except ConfigException:
178+ k8s_config.load_kube_config()
179+ 
180+ await asyncio.to_thread(_load)
181+ self._config_loaded = True
182+ 
183+ async def _get_apis(self) -> tuple[client.CoreV1Api, client.AppsV1Api]:
184+ await self._ensure_client_config()
185+ return client.CoreV1Api(), client.AppsV1Api()
186+ 
187+ def _resolve_ir_file(self, ctx: DeployContext[K8sParams]) -> tuple[str, str]:
188+ k8s_params = ctx.params or K8sParams()
189+ ir_path = getattr(k8s_params, "ir_path", None)
190+ if not ir_path:
191+ raise RuntimeError("ir_path is required for k8s deployment")
192+ 
193+ source_path = Path(ir_path)
194+ if not source_path.exists():
195+ raise RuntimeError(f"ir_path not found: {ir_path}")
196+ return source_path.name, source_path.read_text(encoding="utf-8")
197+ 
198+ def _resolve_runtime_settings(
199+ self, ctx: DeployContext[K8sParams]
200+ ) -> RuntimeSettings:
201+ k8s_params = ctx.params or K8sParams()
202+ deployment_conf = k8s_params.deployment
203+ service_conf = k8s_params.service
204+ 
205+ namespace = k8s_params.namespace or self.namespace
206+ deployment_name = k8s_params.deployment_name or self._default_resource_name(ctx.deployment_id)
207+ replicas = deployment_conf.replicas
208+ 
209+ container_conf = deployment_conf.container
210+ image = container_conf.image
211+ if not image:
212+ raise RuntimeError("image is required for k8s deployment")
213+ image_pull_policy = container_conf.image_pull_policy
214+ container_port = (
215+ container_conf.container_port
216+ or ctx.port
217+ )
218+ service_port = (
219+ service_conf.service_port
220+ or container_port
221+ or ctx.port
222+ )
223+ service_type = service_conf.service_type or "LoadBalancer"
224+ node_port = service_conf.node_port
225+ node_selector = self._normalize_mapping(deployment_conf.node_selector)
226+ userdata = getattr(k8s_params, "userdata", None)
227+ 
228+ return RuntimeSettings(
229+ namespace=namespace,
230+ deployment_name=deployment_name,
231+ replicas=replicas,
232+ container_port=container_port,
233+ service_port=service_port,
234+ service_type=service_type,
235+ node_port=node_port,
236+ node_selector=node_selector,
237+ image=image,
238+ image_pull_policy=image_pull_policy,
239+ userdata=userdata,
240+ )
241+ 
242+ def _build_secret_body(
243+ self,
244+ *,
245+ name: str,
246+ namespace: str,
247+ labels: dict[str, Any],
248+ config_file_name: str,
249+ config_content: str,
250+ ) -> client.V1Secret:
251+ return client.V1Secret(
252+ api_version="v1",
253+ kind="Secret",
254+ metadata=client.V1ObjectMeta(
255+ name=f"{name}-config",
256+ namespace=namespace,
257+ labels={key: str(value) for key, value in labels.items()},
258+ ),
259+ type="Opaque",
260+ string_data={config_file_name: config_content},
261+ )
262+ 
263+ def _build_deployment_body(
264+ self,
265+ *,
266+ name: str,
267+ namespace: str,
268+ labels: dict[str, Any],
269+ replicas: int,
270+ image: str,
271+ image_pull_policy: str,
272+ container_port: int,
273+ config_file_name: str,
274+ node_selector: Optional[dict[str, Any]],
275+ env_vars: dict[str, Any],
276+ ) -> client.V1Deployment:
277+ container = client.V1Container(
278+ name=name,
279+ image=image,
280+ image_pull_policy=image_pull_policy,
281+ command=["python"],
282+ args=[
283+ "-m",
284+ "openjiuwen_runtime.examples.lowcode_agent",
285+ "--irpath",
286+ f"/app/config/{config_file_name}",
287+ "--host",
288+ "0.0.0.0",
289+ "--port",
290+ str(container_port),
291+ ],
292+ env=[
293+ client.V1EnvVar(name=key, value=str(value))
294+ for key, value in env_vars.items()
295+ if value is not None
296+ ],
297+ ports=[
298+ client.V1ContainerPort(
299+ name="http",
300+ container_port=container_port,
301+ )
302+ ],
303+ volume_mounts=[
304+ client.V1VolumeMount(
305+ name="agent-config",
306+ mount_path="/app/config",
307+ read_only=True,
308+ )
309+ ],
310+ startup_probe=client.V1Probe(
311+ http_get=client.V1HTTPGetAction(path="/health", port="http"),
312+ failure_threshold=30,
313+ period_seconds=10,
314+ ),
315+ readiness_probe=client.V1Probe(
316+ http_get=client.V1HTTPGetAction(path="/health", port="http"),
317+ initial_delay_seconds=5,
318+ period_seconds=10,
319+ ),
320+ liveness_probe=client.V1Probe(
321+ http_get=client.V1HTTPGetAction(path="/health", port="http"),
322+ initial_delay_seconds=30,
323+ period_seconds=20,
324+ ),
325+ )
326+ 
327+ pod_spec = client.V1PodSpec(
328+ node_selector={key: str(value) for key, value in (node_selector or {}).items()} or None,
329+ containers=[container],
330+ volumes=[
331+ client.V1Volume(
332+ name="agent-config",
333+ secret=client.V1SecretVolumeSource(
334+ secret_name=f"{name}-config",
335+ items=[
336+ client.V1KeyToPath(
337+ key=config_file_name,
338+ path=config_file_name,
339+ )
340+ ],
341+ ),
342+ )
343+ ],
344+ )
345+ 
346+ template = client.V1PodTemplateSpec(
347+ metadata=client.V1ObjectMeta(labels={key: str(value) for key, value in labels.items()}),
348+ spec=pod_spec,
349+ )
350+ 
351+ spec = client.V1DeploymentSpec(
352+ replicas=replicas,
353+ selector=client.V1LabelSelector(match_labels={"app": name}),
354+ template=template,
355+ )
356+ 
357+ return client.V1Deployment(
358+ api_version="apps/v1",
359+ kind="Deployment",
360+ metadata=client.V1ObjectMeta(
361+ name=name,
362+ namespace=namespace,
363+ labels={key: str(value) for key, value in labels.items()},
364+ ),
365+ spec=spec,
366+ )
367+ 
368+ def _build_service_body(
369+ self,
370+ *,
371+ name: str,
372+ namespace: str,
373+ labels: dict[str, Any],
374+ service_type: str,
375+ service_port: int,
376+ node_port: Optional[int],
377+ ) -> client.V1Service:
378+ service_port_body = client.V1ServicePort(
379+ name="http",
380+ protocol="TCP",
381+ port=service_port,
382+ target_port="http",
383+ node_port=int(node_port) if node_port is not None else None,
384+ )
385+ spec = client.V1ServiceSpec(
386+ type=service_type,
387+ selector={"app": name},
388+ ports=[service_port_body],
389+ )
390+ return client.V1Service(
391+ api_version="v1",
392+ kind="Service",
393+ metadata=client.V1ObjectMeta(
394+ name=name,
395+ namespace=namespace,
396+ labels={key: str(value) for key, value in labels.items()},
397+ ),
398+ spec=spec,
399+ )
400+ 
401+ async def _create_or_patch_secret(
402+ self, core_api: client.CoreV1Api, namespace: str, body: client.V1Secret
403+ ):
404+ try:
405+ return await self._call_api(core_api.create_namespaced_secret, namespace=namespace, body=body)
406+ except ApiException as exc:
407+ if not self._is_conflict(exc):
408+ raise
409+ return await self._call_api(
410+ core_api.patch_namespaced_secret,
411+ name=body.metadata.name,
412+ namespace=namespace,
413+ body=body,
414+ )
415+ 
416+ async def _create_or_patch_deployment(
417+ self, apps_api: client.AppsV1Api, namespace: str, body: client.V1Deployment
418+ ):
419+ try:
420+ return await self._call_api(apps_api.create_namespaced_deployment, namespace=namespace, body=body)
421+ except ApiException as exc:
422+ if not self._is_conflict(exc):
423+ raise
424+ return await self._call_api(
425+ apps_api.patch_namespaced_deployment,
426+ name=body.metadata.name,
427+ namespace=namespace,
428+ body=body,
429+ )
430+ 
431+ async def _create_or_patch_service(
432+ self, core_api: client.CoreV1Api, namespace: str, body: client.V1Service
433+ ):
434+ try:
435+ return await self._call_api(core_api.create_namespaced_service, namespace=namespace, body=body)
436+ except ApiException as exc:
437+ if not self._is_conflict(exc):
438+ raise
439+ return await self._call_api(
440+ core_api.patch_namespaced_service,
441+ name=body.metadata.name,
442+ namespace=namespace,
443+ body=body,
444+ )
445+ 
446+ async def _wait_for_rollout(
447+ self, apps_api: client.AppsV1Api, namespace: str, deployment_name: str, replicas: int
448+ ) -> tuple[bool, str]:
449+ deadline = asyncio.get_running_loop().time() + self.rollout_timeout
450+ while True:
451+ deployment = await self._call_api(
452+ apps_api.read_namespaced_deployment,
453+ name=deployment_name,
454+ namespace=namespace,
455+ )
456+ status = deployment.status or client.V1DeploymentStatus()
457+ generation = deployment.metadata.generation or 0
458+ observed_generation = status.observed_generation or 0
459+ ready_replicas = status.ready_replicas or 0
460+ available_replicas = status.available_replicas or 0
461+ 
462+ if (
463+ observed_generation >= generation
464+ and ready_replicas >= replicas
465+ and available_replicas >= replicas
466+ ):
467+ return True, f"deployment/{deployment_name} successfully rolled out"
468+ 
469+ if asyncio.get_running_loop().time() >= deadline:
470+ return False, f"Timed out waiting for deployment/{deployment_name} rollout"
471+ await asyncio.sleep(self.rollout_poll_interval)
472+ 
473+ async def _wait_for_deployment_deleted(
474+ self,
475+ apps_api: client.AppsV1Api,
476+ namespace: str,
477+ deployment_name: str,
478+ timeout: int,
479+ ) -> tuple[bool, str]:
480+ deadline = asyncio.get_running_loop().time() + timeout
481+ while True:
482+ try:
483+ await self._call_api(
484+ apps_api.read_namespaced_deployment,
485+ name=deployment_name,
486+ namespace=namespace,
487+ )
488+ except ApiException as exc:
489+ if self._is_not_found(exc):
490+ return True, f"deployment/{deployment_name} deleted"
491+ return False, self._format_api_exception(exc)
492+ 
493+ if asyncio.get_running_loop().time() >= deadline:
494+ return False, f"Timed out waiting for deployment/{deployment_name} deletion"
495+ await asyncio.sleep(self.rollout_poll_interval)
496+ 
497+ async def _wait_for_pods_deleted(
498+ self,
499+ core_api: client.CoreV1Api,
500+ namespace: str,
501+ deployment_name: str,
502+ timeout: int,
503+ ) -> tuple[bool, str]:
504+ deadline = asyncio.get_running_loop().time() + timeout
505+ label_selector = f"app={deployment_name}"
506+ while True:
507+ pod_list = await self._call_api(
508+ core_api.list_namespaced_pod,
509+ namespace=namespace,
510+ label_selector=label_selector,
511+ )
512+ if not pod_list.items:
513+ return True, f"pods for {deployment_name} deleted"
514+ 
515+ if asyncio.get_running_loop().time() >= deadline:
516+ pod_names = ",".join(
517+ pod.metadata.name for pod in pod_list.items if pod.metadata and pod.metadata.name
518+ )
519+ return False, f"Timed out waiting for pods deletion: {pod_names}"
520+ await asyncio.sleep(self.rollout_poll_interval)
57 521 
58 async def deploy(self, ctx: DeployContext[K8sParams]) -> DeployResult:522 async def deploy(self, ctx: DeployContext[K8sParams]) -> DeployResult:
59- logger.info("Deploying k8s: deployment_id=%s, host=%s", ctx.deployment_id, ctx.host)523+ deployment_id = ctx.deployment_id
524+ logger.info("Deploying k8s with client: deployment_id=%s, host=%s", deployment_id, ctx.host)
60 try:525 try:
61- k8s_params = ctx.params or K8sParams()526+ settings = self._resolve_runtime_settings(ctx)
62- namespace = k8s_params.namespace or self.namespace
63- deployment_name = k8s_params.deployment_name or f"deploy-{ctx.deployment_id}"
64- replicas = k8s_params.replicas or 1
65- config_map = k8s_params.config_map
66- secret = k8s_params.secret
67- host = ctx.host or self.default_host
68 527 
69- if config_map:528+ config_file_name, config_content = self._resolve_ir_file(ctx)
70- logger.debug("Creating config map: namespace=%s", namespace)
71- cm_manifest = self._build_config_map_manifest(deployment_name, namespace, config_map)
72- success, output = await self._run_kubectl_command("apply", "-f", "-", "--namespace", namespace)
73- if not success:
74- logger.error(
75- "ConfigMap creation failed: deployment_id=%s, error=%s",
76- ctx.deployment_id,
77- output,
78- )
79 529 
80- if secret:530+ labels = {
81- logger.debug("Creating secret: namespace=%s", namespace)531+ "app": settings.deployment_name,
82- secret_manifest = self._build_secret_manifest(deployment_name, namespace, secret)532+ "openjiuwen/deployment-id": deployment_id,
83- success, output = await self._run_kubectl_command("apply", "-f", "-", "--namespace", namespace)533+ "openjiuwen/type": "pod",
84- if not success:534+ }
85- logger.error(535+ env_vars = {
86- "Secret creation failed: deployment_id=%s, error=%s",536+ "DEPLOYMENT_ID": deployment_id,
87- ctx.deployment_id,537+ "AGENT_CONFIG_FILE": f"/app/config/{config_file_name}",
88- output,538+ "CONTAINER_PORT": str(settings.container_port),
89- )539+ }
540+ if settings.userdata:
541+ env_vars["RUNTIME_USERDATA"] = settings.userdata
90 542 
91- deployment_manifest = self._build_deployment_manifest(543+ core_api, apps_api = await self._get_apis()
92- deployment_name, namespace, replicas, ctx.deployment_id544+ secret_body = self._build_secret_body(
545+ name=settings.deployment_name,
546+ namespace=settings.namespace,
547+ labels=labels,
548+ config_file_name=config_file_name,
549+ config_content=config_content,
550+ )
551+ deployment_body = self._build_deployment_body(
552+ name=settings.deployment_name,
553+ namespace=settings.namespace,
554+ labels=labels,
555+ replicas=settings.replicas,
556+ image=settings.image,
557+ image_pull_policy=settings.image_pull_policy,
558+ container_port=settings.container_port,
559+ config_file_name=config_file_name,
560+ node_selector=settings.node_selector,
561+ env_vars=env_vars,
562+ )
563+ service_body = self._build_service_body(
564+ name=settings.deployment_name,
565+ namespace=settings.namespace,
566+ labels=labels,
567+ service_type=settings.service_type,
568+ service_port=settings.service_port,
569+ node_port=settings.node_port,
93 )570 )
94 571 
95- logger.debug("Creating deployment: deployment_name=%s", deployment_name)572+ await self._create_or_patch_secret(core_api, settings.namespace, secret_body)
96- success, output = await self._run_kubectl_command(573+ await self._create_or_patch_deployment(apps_api, settings.namespace, deployment_body)
97- "apply", "-f", "-", "--namespace", namespace574+ await self._create_or_patch_service(core_api, settings.namespace, service_body)
98- )
99 575 
576+ self._deployments[deployment_id] = settings.deployment_name
577+ success, output = await self._wait_for_rollout(apps_api, settings.namespace,
578+ settings.deployment_name, settings.replicas)
100 if not success:579 if not success:
101- logger.error(
102- "kubectl apply failed: deployment_id=%s, error=%s",
103- ctx.deployment_id,
104- output,
105- )
106 return DeployResult(580 return DeployResult(
107- success=False, message=f"kubectl apply failed: {output}"581+ success=False,
582+ deployment_id=deployment_id,
583+ message=f"Deployment rollout failed: {output}",
108 )584 )
109 585 
110- self._deployments[ctx.deployment_id] = deployment_name586+ url = ctx.url
587+ if not url:
588+ host = ctx.host
589+ kube_host, kube_port = self._get_kubeconfig_host() or (None, None)
590+ if not host or host in {"localhost", "127.0.0.1"}:
591+ host = kube_host or "127.0.0.1"
592+ if settings.node_port is not None:
593+ expose_port = int(settings.node_port)
594+ elif kube_port is not None:
595+ expose_port = int(kube_port)
596+ else:
597+ expose_port = settings.service_port
598+ url = f"http://{host}:{expose_port}"
599+ elif not url.startswith(("http://", "https://")):
600+ url = f"http://{url}"
111 601 
112- logger.debug("Waiting for rollout: deployment_name=%s", deployment_name)
113- success, output = await self._run_kubectl_command(
114- "rollout", "status", f"deployment/{deployment_name}", "-n", namespace
115- )
116- 
117- if not success:
118- logger.error(
119- "Deployment rollout failed: deployment_id=%s, error=%s",
120- ctx.deployment_id,
121- output,
122- )
123- return DeployResult(
124- success=False, message=f"Deployment rollout failed: {output}"
125- )
126- 
127- url = f"http://{host}"
128- 
129- logger.info(
130- "K8s deployed: deployment_id=%s, deployment_name=%s, url=%s",
131- ctx.deployment_id,
132- deployment_name,
133- url,
134- )
135 return DeployResult(602 return DeployResult(
136 success=True,603 success=True,
604+ deployment_id=deployment_id,
137 message="K8s deployment started successfully",605 message="K8s deployment started successfully",
138 url=url,606 url=url,
139 )607 )
140- 608+ except ApiException as exc:
141- except Exception as e:609+ message = self._format_api_exception(exc)
142- logger.error("K8s deploy failed: deployment_id=%s, error=%s", ctx.deployment_id, str(e))610+ logger.error("K8s deploy failed: deployment_id=%s, error=%s", deployment_id, message)
143- return DeployResult(success=False, message=f"Deployment failed: {str(e)}")611+ return DeployResult(
144- 612+ success=False,
145- def _build_deployment_manifest(self, name: str, namespace: str, replicas: int, deployment_id: str) -> str:613+ deployment_id=deployment_id,
146- return f"""614+ message=f"Deployment failed: {message}",
147-apiVersion: apps/v1615+ )
148-kind: Deployment616+ except Exception as exc:
149-metadata:617+ message = self._format_exception_message(exc)
150- name: {name}618+ logger.error("K8s deploy failed: deployment_id=%s, error=%s", deployment_id, message)
151- namespace: {namespace}619+ return DeployResult(
152-spec:620+ success=False,
153- replicas: {replicas}621+ deployment_id=deployment_id,
154- selector:622+ message=f"Deployment failed: {message}",
155- matchLabels:623+ )
156- app: {name}
157- template:
158- metadata:
159- labels:
160- app: {name}
161- spec:
162- containers:
163- - name: {name}
164- image: {deployment_id}:latest
165- ports:
166- - containerPort: 8000
167-"""
168- 
169- def _build_config_map_manifest(self, name: str, namespace: str, data: dict) -> str:
170- import json
171- return f"""
172-apiVersion: v1
173-kind: ConfigMap
174-metadata:
175- name: {name}-config
176- namespace: {namespace}
177-data:
178- {json.dumps(data, indent=2)}
179-"""
180- 
181- def _build_secret_manifest(self, name: str, namespace: str, data: dict) -> str:
182- import json
183- import base64
184- encoded_data = {k: base64.b64encode(v.encode()).decode() for k, v in data.items()}
185- return f"""
186-apiVersion: v1
187-kind: Secret
188-metadata:
189- name: {name}-secret
190- namespace: {namespace}
191-type: Opaque
192-data:
193- {json.dumps(encoded_data, indent=2)}
194-"""
195 624 
196 async def stop(self, deployment_id: str, **kwargs) -> DeployResult:625 async def stop(self, deployment_id: str, **kwargs) -> DeployResult:
197- logger.info("Stopping k8s: deployment_id=%s", deployment_id)626+ logger.info("Stopping k8s with client: deployment_id=%s", deployment_id)
627+ namespace = kwargs.get("namespace") or self.namespace
628+ deployment_name = kwargs.get("deployment_name") or self._default_resource_name(deployment_id)
629+ timeout = int(kwargs.get("timeout") or 60)
630+ grace_period = int(kwargs.get("grace_period_seconds") or 30)
198 try:631 try:
199- deployment_name = f"deploy-{deployment_id}"632+ core_api, apps_api = await self._get_apis()
200 633 
201- logger.debug("Deleting deployment: deployment_name=%s", deployment_name)634+ try:
202- success, output = await self._run_kubectl_command(635+ await self._call_api(
203- "delete", "deployment", deployment_name, "-n", self.namespace636+ apps_api.delete_namespaced_deployment,
204- )637+ name=deployment_name,
205- 638+ namespace=namespace,
206- if not success:639+ body=client.V1DeleteOptions(
207- logger.error(640+ grace_period_seconds=grace_period,
208- "kubectl delete failed: deployment_id=%s, error=%s",641+ propagation_policy="Foreground",
209- deployment_id,642+ ),
210- output,
211 )643 )
644+ logger.info(
645+ "deployment/%s deletion initiated with %ss grace period",
646+ deployment_name,
647+ grace_period,
648+ )
649+ except ApiException as exc:
650+ if not self._is_not_found(exc):
651+ raise
652+ logger.warning("deployment/%s not found", deployment_name)
653+ 
654+ deployment_deleted, deployment_message = await self._wait_for_deployment_deleted(
655+ apps_api,
656+ namespace,
657+ deployment_name,
658+ timeout,
659+ )
660+ if not deployment_deleted:
212 return DeployResult(661 return DeployResult(
213- success=False, message=f"kubectl delete failed: {output}"662+ success=False,
663+ deployment_id=deployment_id,
664+ message=f"Stop failed: {deployment_message}",
214 )665 )
215 666 
216- if deployment_id in self._deployments:667+ pods_deleted, pods_message = await self._wait_for_pods_deleted(
217- del self._deployments[deployment_id]668+ core_api,
218- 669+ namespace,
219- logger.info("K8s stopped: deployment_id=%s", deployment_id)670+ deployment_name,
220- return DeployResult(671+ timeout,
221- success=True, message="K8s deployment stopped successfully"
222 )672 )
673+ if not pods_deleted:
674+ return DeployResult(
675+ success=False,
676+ deployment_id=deployment_id,
677+ message=f"Stop failed: {pods_message}",
678+ )
223 679 
224- except Exception as e:680+ for resource_name, delete_call in [
225- logger.error("K8s stop failed: deployment_id=%s, error=%s", deployment_id, str(e))681+ (
226- return DeployResult(success=False, message=f"Stop failed: {str(e)}")682+ f"service/{deployment_name}",
683+ lambda: core_api.delete_namespaced_service(
684+ name=deployment_name,
685+ namespace=namespace,
686+ body=client.V1DeleteOptions(),
687+ ),
688+ ),
689+ (
690+ f"secret/{deployment_name}-config",
691+ lambda: core_api.delete_namespaced_secret(
692+ name=f"{deployment_name}-config",
693+ namespace=namespace,
694+ body=client.V1DeleteOptions(),
695+ ),
696+ ),
697+ ]:
698+ try:
699+ await self._call_api(delete_call)
700+ logger.info("%s deletion initiated", resource_name)
701+ except ApiException as exc:
702+ if not self._is_not_found(exc):
703+ raise
704+ logger.warning("%s not found", resource_name)
705+ 
706+ self._deployments.pop(deployment_id, None)
707+ return DeployResult(
708+ success=True,
709+ deployment_id=deployment_id,
710+ message=f"K8s deployment stopped successfully: {deployment_message}; {pods_message}",
711+ )
712+ except ApiException as exc:
713+ message = self._format_api_exception(exc)
714+ logger.error("K8s stop failed: deployment_id=%s, error=%s", deployment_id, message)
715+ return DeployResult(
716+ success=False,
717+ deployment_id=deployment_id,
718+ message=f"Stop failed: {message}",
719+ )
227 720 
228 async def get_status(self, deployment_id: str, **kwargs) -> DeploymentStatus:721 async def get_status(self, deployment_id: str, **kwargs) -> DeploymentStatus:
229- logger.debug("Getting k8s status: deployment_id=%s", deployment_id)722+ """获取 K8s 部署状态(通过 Pod phase 判断)
230- deployment_name = f"deploy-{deployment_id}"
231 723 
232- success, output = await self._run_kubectl_command(724+ Args:
233- "get", "deployment", deployment_name, "-n", self.namespace, "-o",725+ deployment_id: 部署ID
234- "jsonpath={.status.readyReplicas}"726+ **kwargs: namespace, deployment_name
727+ 
728+ Returns:
729+ DeploymentStatus: 部署状态
730+ """
731+ namespace = kwargs.get("namespace") or self.namespace
732+ deployment_name = (
733+ kwargs.get("deployment_name")
734+ or self._deployments.get(deployment_id)
735+ or self._default_resource_name(deployment_id)
235 )736 )
737+ logger.debug("Getting k8s status: deployment_id=%s, deployment_name=%s", deployment_id, deployment_name)
236 738 
237- if not success:739+ try:
740+ core_api, _ = await self._get_apis()
741+ pod_list = await self._call_api(
742+ core_api.list_namespaced_pod,
743+ namespace=namespace,
744+ label_selector=f"app={deployment_name}",
745+ )
746+ except ApiException as exc:
747+ logger.warning("Failed to list pods for %s: %s", deployment_name, self._format_api_exception(exc))
748+ return DeploymentStatus.PENDING
749+ 
750+ if not pod_list.items:
238 return DeploymentStatus.STOPPED751 return DeploymentStatus.STOPPED
239 752 
240- if output and output != "0":753+ has_running_not_ready = False
241- return DeploymentStatus.RUNNING754+ for pod in pod_list.items:
242- else:755+ pod_info = pod.to_dict() if pod else None
243- return DeploymentStatus.PENDING756+ if not pod_info or "status" not in pod_info:
757+ continue
758+ phase = (pod_info["status"].get("phase") or "").lower()
759+ if phase in ("failed", "succeeded"):
760+ return DeploymentStatus.STOPPED
761+ if phase == "running":
762+ container_statuses = pod_info["status"].get("container_statuses") or []
763+ all_ready = all(cs.get("ready", False) for cs in container_statuses) and container_statuses
764+ if all_ready:
765+ return DeploymentStatus.RUNNING
766+ has_running_not_ready = True
767+ 
768+ if has_running_not_ready:
769+ return DeploymentStatus.RUNNING_NOTREADY
770+ return DeploymentStatus.PENDING
@@ -1,25 +1,47 @@
1-# coding: utf-8
2-# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3- 
4"""K8s 部署模块数据模型"""1"""K8s 部署模块数据模型"""
5 2 
6from dataclasses import dataclass, field3from dataclasses import dataclass, field
7from datetime import datetime4from datetime import datetime
8-from typing import Any, Optional5+from typing import Any, Optional, List
9 6 
10from pydantic import BaseModel, Field7from pydantic import BaseModel, Field
11 8 
12from openjiuwen_runtime.foundation.db.table_def import TableDefinition, ColumnDefinition, IndexDefinition9from openjiuwen_runtime.foundation.db.table_def import TableDefinition, ColumnDefinition, IndexDefinition
13 10 
14 11 
12+class K8sContainer(BaseModel):
13+ image: Optional[str] = Field(None, description="镜像")
14+ image_pull_policy: Optional[str] = Field("IfNotPresent", description="镜像下载规则")
15+ container_port: Optional[int] = None
16+ 
17+ 
18+class K8sDeployment(BaseModel):
19+ replicas: Optional[int] = Field(1, description="副本数")
20+ container: Optional[K8sContainer] = None
21+ node_selector: Optional[dict] = None
22+ deployment_type: Optional[str] = Field("pod", description="部署类型")
23+ 
24+ 
25+class K8sService(BaseModel):
26+ service_type: Optional[str] = Field("LoadBalancer", description="服务规则")
27+ node_port: Optional[int] = None
28+ target_port: Optional[int] = None
29+ service_port: Optional[int] = None
30+ 
31+ 
15@dataclass32@dataclass
16class K8sParams:33class K8sParams:
17 """K8s 部署参数"""34 """K8s 部署参数"""
18 namespace: Optional[str] = None35 namespace: Optional[str] = None
19 deployment_name: Optional[str] = None36 deployment_name: Optional[str] = None
20- replicas: Optional[int] = None
21 config_map: Optional[dict[str, Any]] = None37 config_map: Optional[dict[str, Any]] = None
22 secret: Optional[dict[str, str]] = None38 secret: Optional[dict[str, str]] = None
39+ whl_path: Optional[str] = None
40+ package_name: Optional[str] = None
41+ deployment: Optional[K8sDeployment] = field(default=None)
42+ service: Optional[K8sService] = field(default=None)
43+ ir_path: Optional[str] = None
44+ userdata: Optional[str] = None
23 45 
24 46 
25class K8sInfo(BaseModel):47class K8sInfo(BaseModel):
@@ -28,11 +50,17 @@ class K8sInfo(BaseModel):
28 deployment_id: str50 deployment_id: str
29 version: str51 version: str
30 host: str52 host: str
53+ port: Optional[int] = None
31 url: Optional[str] = None54 url: Optional[str] = None
32- pid: Optional[int] = None
33 whl_path: Optional[str] = None55 whl_path: Optional[str] = None
56+ ir_path: Optional[str] = None
34 package_name: Optional[str] = None57 package_name: Optional[str] = None
35- template_file: Optional[str] = None58+ namespace: Optional[str] = None
59+ deployment_name: Optional[str] = None
60+ replicas: Optional[int] = None
61+ image: Optional[str] = None
62+ container_port: Optional[int] = None
63+ node_port: Optional[int] = None
36 created_at: Optional[datetime] = None64 created_at: Optional[datetime] = None
37 updated_at: Optional[datetime] = None65 updated_at: Optional[datetime] = None
38 data: Optional[dict[str, Any]] = None66 data: Optional[dict[str, Any]] = None
@@ -43,16 +71,27 @@ class K8sInfo(BaseModel):
43 71 
44class K8sCreate(BaseModel):72class K8sCreate(BaseModel):
45 """创建 K8s 部署请求模型"""73 """创建 K8s 部署请求模型"""
46- deployment_id: str74+ deployment_id: str = Field(..., description="部署ID")
47- version: str75+ version: str = Field(..., description="版本号")
48- host: str76+ host: str = Field(..., description="主机地址")
49- url: Optional[str] = None77+ port: Optional[int] = Field(None, description="端口")
50- pid: Optional[int] = None78+ url: Optional[str] = Field(None, description="服务URL")
51- whl_path: Optional[str] = None79+ whl_path: Optional[str] = Field(None, description="WHL包路径")
52- package_name: Optional[str] = None80+ ir_path: Optional[str] = Field(None, description="IR文件路径")
53- template_file: Optional[str] = None81+ package_name: Optional[str] = Field(None, description="包名称")
54- data: Optional[dict[str, Any]] = None82+ namespace: Optional[str] = Field(None, description="命名空间")
55- 83+ deployment_name: Optional[str] = Field(None, description="部署名称")
84+ replicas: Optional[int] = Field(1, description="副本数")
85+ node_selector: Optional[dict[str, Any]] = Field(None, description="选择部署节点")
86+ deployment_type: Optional[str] = Field(None, description="部署类型")
87+ image: Optional[str] = Field(None, description="镜像")
88+ image_pull_policy: Optional[str] = Field(None, description="镜像拉取规则")
89+ container_port: Optional[int] = Field(None, description="容器端口")
90+ service_type: Optional[str] = Field(None, description="服务类型")
91+ service_port: Optional[int] = Field(None, description="服务端口")
92+ node_port: Optional[int] = Field(None, description="对外访问端口")
93+ target_port: Optional[int] = Field(None, description="目标端口")
94+ data: Optional[dict[str, Any]] = Field(None, description="扩展数据")
56 95 
57K8S_TABLE_DEF = TableDefinition(96K8S_TABLE_DEF = TableDefinition(
58 table_name="k8s",97 table_name="k8s",
@@ -61,16 +100,29 @@ K8S_TABLE_DEF = TableDefinition(
61 ColumnDefinition("deployment_id", "string", length=64, unique=True, nullable=False),100 ColumnDefinition("deployment_id", "string", length=64, unique=True, nullable=False),
62 ColumnDefinition("version", "string", length=32, nullable=False),101 ColumnDefinition("version", "string", length=32, nullable=False),
63 ColumnDefinition("host", "string", length=255, nullable=False),102 ColumnDefinition("host", "string", length=255, nullable=False),
103+ ColumnDefinition("port", "integer", nullable=True),
64 ColumnDefinition("url", "string", length=512, nullable=True),104 ColumnDefinition("url", "string", length=512, nullable=True),
65- ColumnDefinition("pid", "integer", nullable=True),
66 ColumnDefinition("whl_path", "string", length=512, nullable=True),105 ColumnDefinition("whl_path", "string", length=512, nullable=True),
106+ ColumnDefinition("ir_path", "string", length=512, nullable=True),
67 ColumnDefinition("package_name", "string", length=255, nullable=True),107 ColumnDefinition("package_name", "string", length=255, nullable=True),
68- ColumnDefinition("template_file", "string", length=512, nullable=True),108+ ColumnDefinition("namespace", "string", length=128, nullable=True),
109+ ColumnDefinition("deployment_name", "string", length=255, nullable=True),
110+ ColumnDefinition("replicas", "integer", nullable=True),
111+ ColumnDefinition("node_selector", "json", nullable=True),
112+ ColumnDefinition("deployment_type", "string", length=255, nullable=True),
113+ ColumnDefinition("image", "string", length=512, nullable=True),
114+ ColumnDefinition("image_pull_policy", "string", length=512, nullable=True),
115+ ColumnDefinition("container_port", "integer", nullable=True),
116+ ColumnDefinition("service_type", "string", length=512, nullable=True),
117+ ColumnDefinition("service_port", "integer", nullable=True),
118+ ColumnDefinition("node_port", "integer", nullable=True),
119+ ColumnDefinition("target_port", "integer", nullable=True),
69 ColumnDefinition("created_at", "datetime", nullable=False),120 ColumnDefinition("created_at", "datetime", nullable=False),
70 ColumnDefinition("updated_at", "datetime", nullable=False),121 ColumnDefinition("updated_at", "datetime", nullable=False),
71 ColumnDefinition("data", "json", nullable=True),122 ColumnDefinition("data", "json", nullable=True),
72 ],123 ],
73 indexes=[124 indexes=[
74 IndexDefinition(["deployment_id"], unique=True),125 IndexDefinition(["deployment_id"], unique=True),
126+ IndexDefinition(["namespace", "deployment_name"], unique=False),
75 ],127 ],
76)128)
@@ -1,20 +1,60 @@
1-# coding: utf-8
2-# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3- 
4"""K8s 部署策略"""1"""K8s 部署策略"""
5 2 
3+import json
4+import re
6from datetime import datetime5from datetime import datetime
7from typing import Any6from typing import Any
8 7 
9-from .deployer import K8sDeployer8+from openjiuwen_runtime.foundation.config import settings
10-from .models import K8sInfo, K8sParams, K8S_TABLE_DEF9+from openjiuwen_runtime.foundation.log import get_logger
10+ 
11from ..base.models import DeployContext, CommonParams11from ..base.models import DeployContext, CommonParams
12from ..base.strategy import BaseDeploymentStrategy12from ..base.strategy import BaseDeploymentStrategy
13+from .deployer import K8sDeployer
14+from .models import (
15+ K8sInfo, K8sParams, K8sContainer, K8sDeployment, K8sService, K8S_TABLE_DEF,
16+)
17+ 
18+logger = get_logger(__name__)
19+ 
20+ 
21+def _parse_userdata(raw_data: Any) -> dict:
22+ """从 data 字段中解析 userdata,返回 k8s 配置字典。"""
23+ def _loads_dict(raw_json: str) -> dict:
24+ parsed = json.loads(raw_json)
25+ return parsed if isinstance(parsed, dict) else {}
26+ 
27+ if not isinstance(raw_data, dict):
28+ return {}
29+ userdata = raw_data.get("userdata")
30+ if userdata is None:
31+ return {}
32+ if isinstance(userdata, dict):
33+ return userdata
34+ if isinstance(userdata, str):
35+ try:
36+ return _loads_dict(userdata)
37+ except (json.JSONDecodeError, TypeError):
38+ sanitized = re.sub(r",\s*([}\]])", r"\1", userdata)
39+ if sanitized == userdata:
40+ return {}
41+ try:
42+ logger.warning("Sanitized malformed k8s userdata JSON before parsing")
43+ return _loads_dict(sanitized)
44+ except (json.JSONDecodeError, TypeError):
45+ return {}
46+ return {}
13 47 
14 48 
15class K8sStrategy(BaseDeploymentStrategy[K8sInfo]):49class K8sStrategy(BaseDeploymentStrategy[K8sInfo]):
16 """K8s 部署策略"""50 """K8s 部署策略"""
17 51 
52+ @staticmethod
53+ def _record_to_dict(record: Any) -> dict[str, Any]:
54+ if hasattr(record, "to_dict"):
55+ return record.to_dict()
56+ return record
57+ 
18 def _create_default_deployer(self) -> K8sDeployer:58 def _create_default_deployer(self) -> K8sDeployer:
19 return K8sDeployer()59 return K8sDeployer()
20 60 
@@ -28,43 +68,113 @@ class K8sStrategy(BaseDeploymentStrategy[K8sInfo]):
28 self, deployment_id: str, version: str, **kwargs: Any68 self, deployment_id: str, version: str, **kwargs: Any
29 ) -> dict:69 ) -> dict:
30 now = datetime.utcnow()70 now = datetime.utcnow()
71+ whl_path = kwargs.get("whl_path")
72+ package_name = kwargs.get("package_name")
73+ 
74+ if not package_name and whl_path:
75+ from pathlib import Path
76+ whl_name = Path(whl_path).stem.split("-")[0].lower().replace("_", "-")
77+ package_name = whl_name
78+ 
79+ k8s_defaults = settings.get_k8s_defaults()
80+ k8s_userdata = _parse_userdata(kwargs.get("data"))
81+ 
82+ k8s_cof = {**k8s_defaults, **{k: v for k, v in k8s_userdata.items() if v is not None}}
83+ 
84+ raw_data = kwargs.get("data") or {}
85+ raw_userdata = raw_data.get("userdata") if isinstance(raw_data, dict) else None
86+ record_data = k8s_cof.get("data") or {}
87+ if not isinstance(record_data, dict):
88+ record_data = {}
89+ if raw_userdata is not None:
90+ record_data["userdata"] = raw_userdata
91+ 
31 return {92 return {
32 "deployment_id": deployment_id,93 "deployment_id": deployment_id,
33 "version": version,94 "version": version,
34 "host": kwargs.get("host", "localhost"),95 "host": kwargs.get("host", "localhost"),
96+ "port": kwargs.get("port"),
35 "url": kwargs.get("url"),97 "url": kwargs.get("url"),
36- "pid": kwargs.get("pid"),98+ "whl_path": whl_path,
37- "whl_path": kwargs.get("whl_path"),99+ "ir_path": kwargs.get("ir_path"),
38- "package_name": kwargs.get("package_name"),100+ "package_name": package_name,
39- "template_file": kwargs.get("template_file"),101+ "namespace": k8s_cof.get("namespace", "default"),
102+ "deployment_name": k8s_cof.get("deployment_name"),
103+ "replicas": k8s_cof.get("replicas", 1),
104+ "node_selector": k8s_cof.get("node_selector"),
105+ "deployment_type": k8s_cof.get("deployment_type"),
106+ "image": k8s_cof.get("image"),
107+ "image_pull_policy": k8s_cof.get("image_pull_policy"),
108+ "container_port": k8s_cof.get("container_port"),
109+ "service_type": k8s_cof.get("service_type"),
110+ "service_port": k8s_cof.get("service_port"),
111+ "node_port": k8s_cof.get("node_port"),
112+ "target_port": k8s_cof.get("target_port"),
40 "created_at": now,113 "created_at": now,
41 "updated_at": now,114 "updated_at": now,
42- "data": kwargs.get("data"),115+ "data": record_data if record_data else None,
43 }116 }
44 117 
45 def _build_deploy_context(self, record: Any, deployment: Any) -> DeployContext[K8sParams]:118 def _build_deploy_context(self, record: Any, deployment: Any) -> DeployContext[K8sParams]:
46- if hasattr(record, "to_dict"):119+ data = self._record_to_dict(record)
47- data = record.to_dict()120+ record_data = data.get("data") or {}
48- else:121+ userdata = record_data.get("userdata") if isinstance(record_data, dict) else None
49- data = record122+ 
123+ container = K8sContainer(
124+ container_port=data.get("container_port"),
125+ image=data.get("image"),
126+ image_pull_policy=data.get("image_pull_policy", "IfNotPresent"),
127+ )
128+ 
129+ deployment_obj = K8sDeployment(
130+ replicas=data.get("replicas") or 1,
131+ container=container,
132+ node_selector=data.get("node_selector"),
133+ deployment_type="pod",
134+ )
135+ 
136+ service_port = data.get("service_port") or data.get("container_port") or data.get("port")
137+ service = K8sService(
138+ service_port=service_port,
139+ service_type=data.get("service_type", "LoadBalancer"),
140+ node_port=data.get("node_port"),
141+ target_port=data.get("container_port"),
142+ )
143+ 
144+ k8sparams = K8sParams(
145+ namespace=data.get("namespace"),
146+ deployment_name=data.get("deployment_name"),
147+ config_map=record_data.get("config_map"),
148+ secret=record_data.get("secret"),
149+ whl_path=data.get("whl_path"),
150+ package_name=data.get("package_name"),
151+ deployment=deployment_obj,
152+ service=service,
153+ ir_path=data.get("ir_path"),
154+ userdata=userdata,
155+ )
156+ 
50 return DeployContext(157 return DeployContext(
51 common=CommonParams(158 common=CommonParams(
52 deployment_id=data.get("deployment_id"),159 deployment_id=data.get("deployment_id"),
53 host=data.get("host"),160 host=data.get("host"),
161+ port=data.get("port"),
54 url=data.get("url"),162 url=data.get("url"),
55 ),163 ),
56- params=K8sParams(164+ params=k8sparams,
57- namespace=data.get("namespace"),165+ data=data,
58- deployment_name=data.get("deployment_name"),
59- replicas=data.get("replicas"),
60- config_map=data.get("config_map"),
61- secret=data.get("secret"),
62- ),
63- data=data.get("data"),
64 )166 )
65 167 
66 def _get_stop_kwargs(self, record: Any) -> dict:168 def _get_stop_kwargs(self, record: Any) -> dict:
67- return {}169+ data = self._record_to_dict(record)
170+ return {
171+ "namespace": data.get("namespace"),
172+ "deployment_name": data.get("deployment_name"),
173+ }
68 174 
69 def _get_status_kwargs(self, record: Any) -> dict:175 def _get_status_kwargs(self, record: Any) -> dict:
70- return {}176+ data = self._record_to_dict(record)
177+ return {
178+ "namespace": data.get("namespace"),
179+ "deployment_name": data.get("deployment_name"),
180+ }
@@ -0,0 +1,38 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+"""Dispatcher 入口类 - 消息分发与调度"""
5+from abc import ABC
6+from dataclasses import dataclass
7+from typing import Any
8+ 
9+from openjiuwen_runtime.foundation.db.handler import DBHandler
10+from openjiuwen_runtime.foundation.log import get_logger
11+ 
12+from .session_manager.models import DispatchHeader, DispatchResult
13+ 
14+logger = get_logger(__name__)
15+ 
16+ 
17+@dataclass
18+class DispatcherConfig:
19+ """Dispatcher 配置类"""
20+ 
21+ db_handler: DBHandler
22+ image: str
23+ max_concurrency_per_pod: int = 200
24+ max_instances: int = 10
25+ namespace: str = "default"
26+ knative_domain: str = "default.example.com"
27+ service_name: str = "jiuwen-agent"
28+ target_port: int = 8000
29+ invoke_path: str = "/invoke"
30+ default_ttl: int = 30
31+ 
32+ 
33+class Server(ABC):
34+ def __init__(self, config: DispatcherConfig):
35+ self.config = config
36+ 
37+ async def dispatch(self, header: DispatchHeader, msg: Any) -> DispatchResult:
38+ pass
@@ -1,5 +1,6 @@
1# coding: utf-81# coding: utf-8
2# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved2# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+from __future__ import annotations
3 4 
4import asyncio5import asyncio
5import uuid6import uuid
@@ -20,10 +21,12 @@ from .deployments import (
20 SubprocessStrategy,21 SubprocessStrategy,
21 DockerStrategy,22 DockerStrategy,
22 K8sStrategy,23 K8sStrategy,
24+ K8S_IMPORT_ERROR,
23)25)
24from .models.deployment_params import (26from .models.deployment_params import (
25 DeployAgentParams,27 DeployAgentParams,
26 DeployPluginParams,28 DeployPluginParams,
29+ DeployImageParams,
27 ListDeploymentsParams,30 ListDeploymentsParams,
28)31)
29from .models.enums import DeployMode, DeploymentType, DeploymentStatus32from .models.enums import DeployMode, DeploymentType, DeploymentStatus
@@ -89,11 +92,19 @@ class DeploymentManager:
89 @staticmethod92 @staticmethod
90 def _create_default_strategies() -> dict[DeployMode, BaseDeploymentStrategy]:93 def _create_default_strategies() -> dict[DeployMode, BaseDeploymentStrategy]:
91 """创建默认策略"""94 """创建默认策略"""
92- return {95+ strategies = {
93 DeployMode.SUBPROCESS: SubprocessStrategy(),96 DeployMode.SUBPROCESS: SubprocessStrategy(),
94 DeployMode.DOCKER: DockerStrategy(),97 DeployMode.DOCKER: DockerStrategy(),
95- DeployMode.K8S: K8sStrategy(),98+ 
96 }99 }
100+ if K8sStrategy is not None:
101+ strategies[DeployMode.K8S] = K8sStrategy()
102+ else:
103+ logger.warning(
104+ "K8s strategy disabled because Kubernetes dependencies could not be imported: %s",
105+ K8S_IMPORT_ERROR,
106+ )
107+ return strategies
97 108 
98 @staticmethod109 @staticmethod
99 def _generate_deployment_id() -> str:110 def _generate_deployment_id() -> str:
@@ -183,6 +194,32 @@ class DeploymentManager:
183 )194 )
184 )195 )
185 196 
197+ async def deploy_image(self, params: DeployImageParams) -> DeploymentInfo:
198+ """部署镜像"""
199+ if params.mode == DeployMode.SUBPROCESS:
200+ raise NotImplementedError("deploy_image is not supported for SUBPROCESS mode")
201+ logger.info(
202+ "Deploying image: name=%s, version=%s, mode=%s, user_id=%s, space_id=%s",
203+ params.name,
204+ params.version,
205+ params.mode,
206+ params.user_id,
207+ params.space_id,
208+ )
209+ extras = dict(params.extras)
210+ extras["image"] = params.image
211+ return await self._deploy(
212+ _DeployExecutionParams(
213+ deployment_type=DeploymentType.IMAGE,
214+ name=params.name,
215+ version=params.version,
216+ mode=params.mode,
217+ user_id=params.user_id,
218+ space_id=params.space_id,
219+ extras=extras,
220+ )
221+ )
222+ 
186 async def list_deployments(self, params: ListDeploymentsParams) -> list[DeploymentInfo]:223 async def list_deployments(self, params: ListDeploymentsParams) -> list[DeploymentInfo]:
187 """列出部署"""224 """列出部署"""
188 logger.debug(225 logger.debug(
@@ -252,7 +289,7 @@ class DeploymentManager:
252 logger.warning("Cannot stop deployment, mode not found: deployment_id=%s", deployment_id)289 logger.warning("Cannot stop deployment, mode not found: deployment_id=%s", deployment_id)
253 return False290 return False
254 291 
255- strategy = self._get_strategy(mode)292+ strategy = self.get_strategy(mode)
256 result = await strategy.stop(deployment_id, self.db_handler)293 result = await strategy.stop(deployment_id, self.db_handler)
257 if result.success:294 if result.success:
258 logger.info("Deployment stopped: deployment_id=%s", deployment_id)295 logger.info("Deployment stopped: deployment_id=%s", deployment_id)
@@ -275,7 +312,7 @@ class DeploymentManager:
275 await self.stop_deployment(deployment_id, mode)312 await self.stop_deployment(deployment_id, mode)
276 313 
277 if mode:314 if mode:
278- strategy = self._get_strategy(mode)315+ strategy = self.get_strategy(mode)
279 await strategy.delete_record(self.db_handler, deployment_id)316 await strategy.delete_record(self.db_handler, deployment_id)
280 317 
281 result = await self.db_handler.delete(318 result = await self.db_handler.delete(
@@ -291,23 +328,27 @@ class DeploymentManager:
291 async def get_process_info(self, deployment_id: str) -> Optional[ProcessInfo]:328 async def get_process_info(self, deployment_id: str) -> Optional[ProcessInfo]:
292 """获取进程部署详情"""329 """获取进程部署详情"""
293 logger.debug("Getting process info: deployment_id=%s", deployment_id)330 logger.debug("Getting process info: deployment_id=%s", deployment_id)
294- strategy = self._get_strategy(DeployMode.SUBPROCESS)331+ strategy = self.get_strategy(DeployMode.SUBPROCESS)
295 return await strategy.get_info(self.db_handler, deployment_id)332 return await strategy.get_info(self.db_handler, deployment_id)
296 333 
297 async def get_docker_info(self, deployment_id: str) -> Optional[DockerInfo]:334 async def get_docker_info(self, deployment_id: str) -> Optional[DockerInfo]:
298 """获取Docker部署详情"""335 """获取Docker部署详情"""
299 logger.debug("Getting docker info: deployment_id=%s", deployment_id)336 logger.debug("Getting docker info: deployment_id=%s", deployment_id)
300- strategy = self._get_strategy(DeployMode.DOCKER)337+ strategy = self.get_strategy(DeployMode.DOCKER)
301 return await strategy.get_info(self.db_handler, deployment_id)338 return await strategy.get_info(self.db_handler, deployment_id)
302 339 
303 async def get_k8s_info(self, deployment_id: str) -> Optional[K8sInfo]:340 async def get_k8s_info(self, deployment_id: str) -> Optional[K8sInfo]:
304 """获取K8S部署详情"""341 """获取K8S部署详情"""
305 logger.debug("Getting k8s info: deployment_id=%s", deployment_id)342 logger.debug("Getting k8s info: deployment_id=%s", deployment_id)
306- strategy = self._get_strategy(DeployMode.K8S)343+ strategy = self.get_strategy(DeployMode.K8S)
307 return await strategy.get_info(self.db_handler, deployment_id)344 return await strategy.get_info(self.db_handler, deployment_id)
308 345 
309- def _get_strategy(self, mode: DeployMode) -> BaseDeploymentStrategy:346+ def get_strategy(self, mode: DeployMode) -> BaseDeploymentStrategy:
310 """获取部署策略"""347 """获取部署策略"""
348+ if mode == DeployMode.K8S and K8S_IMPORT_ERROR is not None:
349+ raise RuntimeError(
350+ f"K8s deployment is unavailable because Kubernetes dependencies failed to import: {K8S_IMPORT_ERROR}"
351+ ) from K8S_IMPORT_ERROR
311 return self._strategies[mode]352 return self._strategies[mode]
312 353 
313 async def _detect_deploy_mode(self, deployment_id: str) -> Optional[DeployMode]:354 async def _detect_deploy_mode(self, deployment_id: str) -> Optional[DeployMode]:
@@ -409,13 +450,15 @@ class DeploymentManager:
409 450 
410 await self.db_handler.create(DEPLOYMENT_TABLE_NAME, deployment_data)451 await self.db_handler.create(DEPLOYMENT_TABLE_NAME, deployment_data)
411 452 
412- strategy = self._get_strategy(params.mode)453+ strategy = self.get_strategy(params.mode)
413 454 
414 try:455 try:
415 await strategy.create_record(456 await strategy.create_record(
416 self.db_handler, deployment_id, params.version, **extras457 self.db_handler, deployment_id, params.version, **extras
417 )458 )
418- await strategy.deploy(deployment_id, self.db_handler)459+ deploy_result = await strategy.deploy(deployment_id, self.db_handler)
460+ if not deploy_result.success:
461+ raise RuntimeError(deploy_result.message or f"Deployment {deployment_id} failed")
419 462 
420 await self._wait_until_deployment_ready(deployment_id)463 await self._wait_until_deployment_ready(deployment_id)
421 logger.info(464 logger.info(
@@ -430,6 +473,7 @@ class DeploymentManager:
430 {DeploymentFields.DEPLOYMENT_ID: deployment_id},473 {DeploymentFields.DEPLOYMENT_ID: deployment_id},
431 {DeploymentFields.DEPLOYMENT_STATUS: DeploymentStatus.FAILED.value},474 {DeploymentFields.DEPLOYMENT_STATUS: DeploymentStatus.FAILED.value},
432 )475 )
476+ raise
433 477 
434 deployment_record = await self.db_handler.get(478 deployment_record = await self.db_handler.get(
435 DEPLOYMENT_TABLE_NAME,479 DEPLOYMENT_TABLE_NAME,
@@ -12,10 +12,16 @@ from .schemas import (
12 DeploymentFields,12 DeploymentFields,
13 DeploymentInfo,13 DeploymentInfo,
14)14)
15+from .deployment_params import (
16+ DeployAgentParams,
17+ DeployPluginParams,
18+ DeployImageParams,
19+ ListDeploymentsParams,
20+)
15 21 
16__all__ = [22__all__ = [
17- "DeploymentType", 23+ "DeploymentType",
18- "DeploymentStatus", 24+ "DeploymentStatus",
19 "DeploymentInfo",25 "DeploymentInfo",
20 "DeploymentCreate",26 "DeploymentCreate",
21 "DEPLOYMENT_TABLE_NAME",27 "DEPLOYMENT_TABLE_NAME",
@@ -23,4 +29,8 @@ __all__ = [
23 "TableDefinition",29 "TableDefinition",
24 "ColumnDefinition",30 "ColumnDefinition",
25 "IndexDefinition",31 "IndexDefinition",
32+ "DeployAgentParams",
33+ "DeployPluginParams",
34+ "DeployImageParams",
35+ "ListDeploymentsParams",
26]36]
@@ -44,3 +44,16 @@ class DeployPluginParams:
44 user_id: Optional[str] = None44 user_id: Optional[str] = None
45 space_id: Optional[str] = None45 space_id: Optional[str] = None
46 extras: dict[str, Any] = field(default_factory=dict)46 extras: dict[str, Any] = field(default_factory=dict)
47+ 
48+ 
49+@dataclass
50+class DeployImageParams:
51+ """部署镜像的参数(策略扩展字段放入 extras,如环境变量、端口等配置)"""
52+ 
53+ image: str
54+ name: str
55+ version: str
56+ mode: DeployMode
57+ user_id: Optional[str] = None
58+ space_id: Optional[str] = None
59+ extras: dict[str, Any] = field(default_factory=dict)
@@ -10,6 +10,7 @@ class DeploymentType(str, Enum):
10 """部署类型"""10 """部署类型"""
11 AGENT = "agent"11 AGENT = "agent"
12 PLUGIN = "plugin"12 PLUGIN = "plugin"
13+ IMAGE = "image"
13 14 
14 15 
15class DeploymentStatus(str, Enum):16class DeploymentStatus(str, Enum):
@@ -18,6 +19,7 @@ class DeploymentStatus(str, Enum):
18 RUNNING = "running"19 RUNNING = "running"
19 STOPPED = "stopped"20 STOPPED = "stopped"
20 FAILED = "failed"21 FAILED = "failed"
22+ RUNNING_NOTREADY = "running_not_ready"
21 23 
22 24 
23class DeployMode(str, Enum):25class DeployMode(str, Enum):
@@ -0,0 +1,217 @@
1+# Session 包软件设计说明
2+ 
3+本文档描述 `openjiuwen_runtime.management.session`**整体架构、模块职责、输入输出与端到端业务流程**,便于实现扩展与排障。
4+ 
5+---
6+ 
7+## 1. 定位与目标
8+ 
9+Session 包在「入口请求 → 多实例、多 Session、可伸缩的后端工作负载(如 K8s Pod)」之间,提供一层**编排**能力:
10+ 
11+- **双队列**调度:用户业务请求与系统内部事件分队列,**系统侧优先**
12+- **单服务实例 = 一个部署单元 + 一条下行通道**(典型为 WebSocket 多路复用 `request_id`)。
13+- **两级并发控制****服务级**(实例内总并行度)+ **Session 内**(同一会话内并行度),二者独立、通过信号量 `acquire` 排队配合。
14+- **Session 亲和**:同一 `session_id` 路由到同一 `service_id`,在 TTL 内可滑动续期;可选策略(如 `chat_id` + `bot_id` 生成稳定 session 键)。
15+- **弹性伸缩**`min_idle` 预热、`max_services` 上限、空闲 `service_ttl` 后缩容;部署抽象(K8s / Docker / 无部署调试用 `NoOp`)。
16+ 
17+---
18+ 
19+## 2. 整体架构
20+ 
21+```mermaid
22+flowchart TB
23+ subgraph 入口
24+ A[Access]
25+ end
26+ subgraph 编排
27+ SM[ServiceManager]
28+ DQ[PriorityDualAsyncQueues]
29+ SR[ServiceRouter]
30+ T[Timer]
31+ end
32+ subgraph 单服务实例
33+ SH[ServiceHandler]
34+ SessH[SessionHandler 每 session 一实例]
35+ CH[IServiceMessageChannel 如 WSServiceMessageChannel]
36+ DC[IDeployController 如 K8sDeployController]
37+ end
38+ subgraph 策略
39+ STR[ISessionStrategy 如 PerChatBotStrategy]
40+ end
41+ subgraph 部署
42+ K8S[K8sServiceHandler / Pod]
43+ end
44+ 
45+ A -->|handle_message: SessionRequestWrapper| SM
46+ A -->|init: strategy, config, session_config| STR
47+ SM --> DQ
48+ SM --> SR
49+ SM -->|new_service, deploy| SH
50+ SH --> CH
51+ SH --> DC
52+ DC --> K8S
53+ SH --> SessH
54+ STR -->|handle_session: ISessionRequest| A
55+ CH -->|下行分片: dispatch_inbound_chunk| SH
56+```
57+ 
58+**层次关系**
59+ 
60+| 层 | 职责 |
61+|----|------|
62+| **Access** | 将 `IRequest` 经策略变成 `ISessionRequest`,封 `SessionRequestWrapper`**只负责入队****消费** `response_queue` 的异步迭代。 |
63+| **ServiceManager** | 双队列、路由、多实例池(`in_use` / `idle`)、bootstrap `min_idle`、autoscale、session TTL 与 service idle 回收。 |
64+| **ServiceHandler** | 单实例:服务级信号量、SessionRouter(子 SessionHandler)、`deploy` / `delete`、经 `invoke_channel` 调通道 `send`。 |
65+| **SessionHandler** | 同 `session_id` 的会话内信号量、调用父级 `invoke_channel`。 |
66+| **IServiceMessageChannel** | 上行业务、下行多路分片、完成时 `on_request_complete` 归还服务级并发。 |
67+| **IDeployController** | 创建/删除后端资源,返回 `PodDeployInfo` 等,供通道 `on_pod_ready` 建链。 |
68+ 
69+---
70+ 
71+## 3. 核心类型与「接口 I/O」
72+ 
73+### 3.1 请求与包装
74+ 
75+| 类型 | 说明 | 主要输入 / 输出 |
76+|------|------|-----------------|
77+| `IRequest` | 业务入口行协议(`request_id` / `chat_id` / `bot_id` / `user_id` / `session_id`)。 | 入:由调用方提供实现;`Access` 在缺省 `request_id` 时包一层 `_AutoIdRequest` 并写回 `wire_dict`(若存在)。 |
78+| `ISessionRequest` | 策略产出的**会话化**请求,含 `session_id``session_concurrency``session_ttl``raw`(原 `IRequest`)。 | 出:`ISessionStrategy.handle_session(IRequest) -> ISessionRequest`。实现:`SessionRequest`。 |
79+| `SessionRequestWrapper` | 一次用户调用对应一个 wrapper:`session_request` + `response_queue` + `cancel` Future。 | 入:Access 创建;`ServiceManager` 入队;通道下行 `put``response_queue``Access``is_completed` 判终态。 |
80+ 
81+### 3.2 控制面接口(摘录)
82+ 
83+| 接口 | 方法 | 含义 |
84+|------|------|------|
85+| `IAccess` | `init(...)` / `send_message(IRequest) -> AsyncIterator` | 初始化并 `start` 服务管理;流式产出来自下游的解析后结果。 |
86+| `IServiceManager` | `init` / `start` / `stop` / `handle_message` / `enqueue_system` | 对 wrapper 入用户队列;系统事件入系统队列。 |
87+| `IServiceHandler` | `handle_message` / `deploy` / `delete` / `remove_session` / 并发与 session 只读属性 | 单实例生命周期与消息处理。 |
88+| `IServiceMessageChannel` | `send(service_id, wrapper, *, response_parser, on_request_complete)` | **上行**一帧 + **下行**在独立接收循环中 `dispatch` + 完成时 `await on_request_complete(rid)`。 |
89+| `IResponseParser` | `request_id` / `is_completed` / `response` 作用于 `dict` 分片 | 多路流式与终态判断。 |
90+| `IDeployController` | `deploy() -> info` / `delete()` / `resource_id` | 与具体运行时解耦。 |
91+ 
92+**回调约定**`on_request_complete(Optional[str])` 为可 `await` 的异步回调,**必须**在单条用户请求在通道侧视为结束后调用,以释放 `ServiceHandler` 的服务级信号量。
93+ 
94+### 3.3 数据模型 `models.py`
95+ 
96+- **`AccessConfig`**:双队列大小、`image``target_port``invoke_path``ws_use_tls`**服务池** `min_idle_services` / `max_services``service_concurrency`**实例空闲回收** `service_ttl``message_timeout``autoscale_interval` 等。
97+- **`SessionConfig`**:单策略生效的 `concurrency`(同 session 内最大并行中请求数)、`ttl`(秒;0 表示不启用 session TTL 计时器)。
98+ 
99+### 3.4 内部事件 `internal_events.py`
100+ 
101+- `ServiceReclaimEvent(service_id: str)`:经 `Timer``service_idle_ttl` 触发,**系统队列**优先消费,执行缩容 `delete`
102+ 
103+### 3.5 错误码 `exception.py`
104+ 
105+-`ServiceManager._fail` 配合,向 `response_queue` 写入 `error_code` + `message` + `completed`;常见 `100001`(资源满)、`100002`(路由/处理异常)。
106+ 
107+---
108+ 
109+## 4. 模块与文件功能一览
110+ 
111+| 文件 | 功能 |
112+|------|------|
113+| `access.py` | `Access`:策略生成 `ISessionRequest`、自动补 `request_id`(多路复用必须)、`handle_message` 入队、从 `response_queue` 流式 `yield``init` 时拉 `ServiceManager.start()`。 |
114+| `service_manager.py` | 双队列消费循环、用户消息独立 task 路由、`_pick_or_create` 亲和/选实例/新 deploy、`_bootstrap_min_idle`、autoscale、session 与 service 空闲计时器、`_fail` 写错误。 |
115+| `service_handler.py` | `ServiceHandler``deploy``on_pod_ready``invoke_channel``channel.send``dispatch_inbound_chunk``request_id` 写回、`delete` 先关通道再删资源。 |
116+| `session_handler.py` | `SessionHandler`:会话内 `BoundedSemaphore` + `invoke_channel`。 |
117+| `dual_queue.py` | 系统优先的 `get()`:先 `drain` 系统队列,再与阻塞用户队列用 `asyncio.wait(FIRST_COMPLETED)`。 |
118+| `router.py` | `ServiceRouter``session_id -> service_id``SessionRouter``request_id -> session_id`(在 ServiceHandler 上用于下行匹配)。 |
119+| `ws_client_channel.py` | `WSServiceMessageChannel``serialize_request_payload` / `wire_dict` 上行业务、`_ensure_connected``on_pod_ready` 中预建链、`send` 内等待 `is_completed``cancel``close`。 |
120+| `k8s_service_handler.py` | `K8sServiceHandler` 创建/等待 Pod Ready、`K8sDeployController` 适配 `IDeployController``PodDeployInfo`。 |
121+| `docker_service_handler.py` | Docker 侧部署信息与实现(如存在)。 |
122+| `runtime.py` | `IDeployController` 协议、`NoOpDeployController` 不调真实部署。 |
123+| `strategies/_base.py` + `per_chat_bot.py` | `BaseSessionStrategy``PerChatBotStrategy``f"{chat_id}::{bot_id}"` 为 session 键。 |
124+| `session_request.py` | `SessionRequest` 实现 `ISessionRequest`。 |
125+| `timer.py` | 抽象 `ITimer` 的调度实现,供 `ServiceManager` arm session/service 空闲计时。 |
126+| `interfaces.py` | 上表各类 Protocol/ABC/别名如 `OnRequestCompleteCallback``IServiceMessageChannel`。 |
127+| `__init__.py` | 对外的公开导出。 |
128+ 
129+---
130+ 
131+## 5. 全链路业务流程
132+ 
133+### 5.1 启动(含预热)
134+ 
135+1. 构造 `ServiceManager`(注入 `IServiceInstanceFactory``PriorityDualAsyncQueues``Timer` 等)。
136+2. `Access.init``ServiceManager.init(response_parser)``ServiceManager.start()`
137+3. `start()` 内部:
138+ - 启动 `_message_loop`(从双队列取项)、`_autoscale_loop`
139+ - `await _bootstrap_min_idle()`:在 `lock` 内循环 `min_idle``max_services`,对每个缺口调用 `_new_deployed()`
140+4. `_new_deployed()``factory.new_service(response_parser)` 得到 `ServiceHandler``await h.deploy()`(K8s 等创建 Pod/容器并 **等 Ready**`WSS``on_pod_ready`**预建 WebSocket**),成功后放入 `idle` 池。
141+ 
142+### 5.2 单次用户请求
143+ 
144+```mermaid
145+sequenceDiagram
146+ participant Client
147+ participant Access
148+ participant SM as ServiceManager
149+ participant SH as ServiceHandler
150+ participant Sess as SessionHandler
151+ participant CH as IServiceMessageChannel
152+ 
153+ Client->>Access: send_message(IRequest)
154+ Access->>Access: strategy.handle_session 得到 ISessionRequest
155+ Access->>SM: handle_message(SessionRequestWrapper)
156+ Note over Access,SM: 仅入队,立即返回
157+ Access->>Access: 循环 response_queue.get()
158+ SM->>SH: _handle_user_request: pick 或 new deploy
159+ SH->>Sess: handle_message(wrapper)
160+ Sess->>SH: invoke_channel(wrapper)
161+ SH->>CH: send -> 上行 + 等下行 is_completed
162+ CH->>SH: dispatch_inbound_chunk -> response_queue.put
163+ Access-->>Client: yield response(data)
164+```
165+ 
166+**要点**
167+ 
168+- **入队与消费解耦**`handle_message` 是异步队列写入;真正路由在 `_message_loop` 起的独立 task 中执行,避免单条长请求阻塞下一条入队(多 session 可并行进入不同实例)。
169+- **亲和路由**`_pick_or_create`):
170+ 1.`ServiceRouter` 已有 `session_id -> service_id` 且实例仍在池 → 复用,必要时从 `idle` 提升到 `in_use`
171+ 2. 否则在 `in_use` / `idle` 中找 `available_concurrency >= 1` 的实例。
172+ 3. 否则若未达 `max_services``await _new_deployed()``in_use`
173+ 4. 达上限仍无位 → `_fail(100001)`
174+ 
175+### 5.3 服务级与 Session 级并发
176+ 
177+- **服务级**`ServiceHandler``asyncio.BoundedSemaphore(total_concurrency)`,在 `invoke_channel` 开头 `acquire`,在 `on_request_complete``release`(无论成功失败通常都会走到回调)。
178+- **Session 级**`SessionHandler``BoundedSemaphore(max_parallel)`,由 `SessionConfig`→策略→`ISessionRequest.session_concurrency` 提供上限。
179+ 
180+**单条请求 = 1 点服务级并发**`_NEED = 1``ServiceManager` 中隐式使用)。
181+ 
182+### 5.4 Session TTL 与实例空闲
183+ 
184+-`session_ttl > 0`:成功处理完一条用户请求后 `Timer``sess:{session_id}` arm;到期 `remove_session` 并清 `ServiceRouter` 映射,若无其它 session 且实例在 `in_use` 则可能回落 `idle` 并 arm `svc:{service_id}``service_idle_ttl`
185+- `service_idle_ttl` 到期 → `ServiceReclaimEvent`**系统**队列,确认无活跃 session / inflight 后 `h.delete()`(先关 WSS 再删 Pod 等)。
186+ 
187+### 5.5 部署与 WSS 建链
188+ 
189+- `ServiceHandler.deploy()``pod_info = await deploy_controller.deploy()`,若非空且通道有 `on_pod_ready``await on_pod_ready(service_id, pod_info)`
190+- `WSServiceMessageChannel`:用 `PodDeployInfo.pod_ip`(或 Docker 的 `host`)与**部署声明的 `port`**`ws://`/`wss://` URL,在 `on_pod_ready``await _ensure_connected()` 完成与业务进程握手(避免「K8s Ready 但首包发不出去」的纯懒连问题)。
191+- **上行体**`serialize_request_payload``IRequest`/`wire_dict` 序列化;`request_id` 非空是 **WSS 多路复用硬条件**`Access` 在缺失时可自动生成 UUID 并补 `wire_dict`
192+ 
193+### 5.6 缩容与停止
194+ 
195+- 空闲回收见上;`ServiceManager.stop()` 标记队列关闭、取消 `_message_loop` / `_autoscale_loop` 与用户路由子 task 等。
196+ 
197+---
198+ 
199+## 6. 设计取舍与扩展点
200+ 
201+| 点 | 说明 |
202+|----|------|
203+| **策略** | 新增 `ISessionStrategy` 可换「按 user 维 session」「仅 request_id 维」等。 |
204+| **通道** | 实现 `IServiceMessageChannel`:除 `send` 外可实现 `bind_handler` / `on_pod_ready` / `close` 以配合 `ServiceHandler`。 |
205+| **部署** | 实现 `IDeployController` 即可接 VM、Nomad 等。 |
206+| **存储** | `AccessConfig.db_handler` 预留,当前核心路径可不落库。 |
207+ 
208+---
209+ 
210+## 7. 与测试 / 可执行样例的对应关系
211+ 
212+- 系统级 Mock:`tests/system_tests/management_session/test_session_sdk.py``NoOpDeploy` + 假通道)。
213+- 真 K8s + WSS:`tests/system_tests/management_session/main_k8s_access.py`,用 `K8sServiceHandler` + `WSServiceMessageChannel` 跑通全链路;业务 JSON 可经由 `WireIRequest``IRequest` 实现上送。
214+ 
215+---
216+ 
217+*文档版本与代码包同步于 `openjiuwen_runtime/management/session/`;若行为与实现不一致,以源码为准。*
@@ -0,0 +1,55 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+from .access import Access
5+from .dual_queue import PriorityDualAsyncQueues
6+from .interfaces import (
7+ IAccess,
8+ IRequest,
9+ IResponseParser,
10+ IServiceHandler,
11+ IServiceInstanceFactory,
12+ IServiceManager,
13+ IServiceMessageChannel,
14+ OnRequestCompleteCallback,
15+ ISessionHandler,
16+ ISessionRequest,
17+ ISessionStrategy,
18+ SessionRequestWrapper,
19+)
20+from .models import AccessConfig, MessagePriority, MessageType, SessionConfig
21+from .runtime import IDeployController, NoOpDeployController
22+from .service_handler import ServiceHandler
23+from .service_manager import ServiceManager
24+from .session_request import SessionRequest
25+from .timer import Timer
26+from .ws_client_channel import WSServiceMessageChannel, serialize_request_payload
27+ 
28+__all__ = (
29+ "Access",
30+ "AccessConfig",
31+ "IDeployController",
32+ "IAccess",
33+ "IRequest",
34+ "IResponseParser",
35+ "IServiceHandler",
36+ "IServiceInstanceFactory",
37+ "IServiceManager",
38+ "IServiceMessageChannel",
39+ "OnRequestCompleteCallback",
40+ "ISessionHandler",
41+ "ISessionRequest",
42+ "ISessionStrategy",
43+ "MessagePriority",
44+ "MessageType",
45+ "NoOpDeployController",
46+ "PriorityDualAsyncQueues",
47+ "ServiceHandler",
48+ "ServiceManager",
49+ "SessionConfig",
50+ "SessionRequest",
51+ "SessionRequestWrapper",
52+ "Timer",
53+ "WSServiceMessageChannel",
54+ "serialize_request_payload",
55+)
@@ -0,0 +1,206 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+"""Access:策略生成 session_id / 并发度 / TTL,交给 ServiceManager(双队列)。"""
5+ 
6+import asyncio
7+import uuid
8+from typing import Any, AsyncIterator, Optional
9+ 
10+from openjiuwen_runtime.foundation.log import get_logger
11+ 
12+from .interfaces import (
13+ IAccess,
14+ IRequest,
15+ IResponseParser,
16+ ISessionStrategy,
17+ IServiceManager,
18+ SessionRequestWrapper, ISessionRequest,
19+)
20+from .models import AccessConfig, SessionConfig
21+ 
22+logger = get_logger(__name__)
23+ 
24+ 
25+class _AutoIdRequest(IRequest):
26+ """覆盖 ``request_id`` 为 Access 生成的 UUID,其余字段透传给原始 ``IRequest``。
27+ 
28+ 用途:当用户 ``IRequest.request_id`` 为空时,Access 自动生成;下游
29+ ``WSServiceMessageChannel`` 依靠 ``request_id`` 做多路复用,否则会拒收。
30+ """
31+ 
32+ def __init__(self, base: IRequest, rid: str) -> None:
33+ self._base = base
34+ self._rid = rid
35+ 
36+ @property
37+ def request_id(self) -> Optional[str]:
38+ return self._rid
39+ 
40+ @property
41+ def chat_id(self) -> Optional[str]:
42+ return self._base.chat_id
43+ 
44+ @property
45+ def bot_id(self) -> Optional[str]:
46+ return self._base.bot_id
47+ 
48+ @property
49+ def user_id(self) -> Optional[str]:
50+ return self._base.user_id
51+ 
52+ @property
53+ def session_id(self) -> Optional[str]:
54+ return self._base.session_id
55+ 
56+ @property
57+ def wire_dict(self) -> Any:
58+ # 让 ws_client_channel._to_jsonable 能拿到含 request_id 的上行字典;
59+ # 若原对象没有 wire_dict,返回 None 走默认序列化分支。
60+ wd = getattr(self._base, "wire_dict", None)
61+ if isinstance(wd, dict):
62+ return wd
63+ return None
64+ 
65+ 
66+class Access(IAccess):
67+ def __init__(self, service_manager: IServiceManager) -> None:
68+ self._service_manager = service_manager
69+ self._strategy: Optional[ISessionStrategy] = None
70+ self._response_parser: Optional[IResponseParser] = None
71+ self._config: Optional[AccessConfig] = None
72+ self._shutdown_done: bool = False
73+ 
74+ async def init(
75+ self,
76+ response_parser: IResponseParser,
77+ config: AccessConfig,
78+ session_config: SessionConfig,
79+ strategy: ISessionStrategy = None,
80+ ) -> None:
81+ # 与入口共享的会话维度:并发与 TTL 写入策略,供 handle_session 填充 ISessionRequest
82+ self._response_parser = response_parser
83+ self._config = config
84+ if strategy:
85+ self._strategy = strategy
86+ self._strategy.configure(concurrency=session_config.concurrency, ttl=session_config.ttl)
87+ await self._service_manager.init(response_parser)
88+ await self._service_manager.start()
89+ logger.info(
90+ "Access 已初始化: user_q=%s sys_q=%s image=%s session_max=%s session_ttl=%s "
91+ "service_concurrency=%s min_idle=%s max=%s port=%s path=%s ws_tls=%s service_ttl=%s",
92+ config.user_queue_size,
93+ config.system_queue_size,
94+ config.image,
95+ session_config.concurrency,
96+ session_config.ttl,
97+ config.service_concurrency,
98+ config.min_idle_services,
99+ config.max_services,
100+ config.target_port,
101+ config.invoke_path,
102+ config.ws_use_tls,
103+ config.service_ttl,
104+ )
105+ logger.debug(
106+ "Access init 完成: message_timeout=%s", getattr(config, "message_timeout", None)
107+ )
108+ 
109+ async def shutdown(self) -> None:
110+ """优雅退出:停 ServiceManager 内全部 asyncio 任务与双队列、取消定时器、delete 已拉起的服务。"""
111+ if self._shutdown_done:
112+ logger.debug("Access shutdown 被忽略(幂等): 已关闭")
113+ return
114+ self._shutdown_done = True
115+ await self._service_manager.stop()
116+ logger.info("Access 已 shutdown")
117+ 
118+ async def update_config(
119+ self, config: AccessConfig, session_config: Optional[SessionConfig] = None
120+ ) -> None:
121+ """运行时热更新配置。存量 session/service 不变,新建的使用新值。"""
122+ self._config = config
123+ await self._service_manager.update_config(
124+ min_idle_services=config.min_idle_services,
125+ max_services=config.max_services,
126+ service_idle_ttl=config.service_ttl,
127+ autoscale_interval=config.autoscale_interval,
128+ )
129+ if session_config and self._strategy:
130+ self._strategy.configure(session_config.concurrency, session_config.ttl)
131+ logger.info("Access 配置已热更新")
132+ 
133+ async def send_message(self, msg: IRequest | ISessionRequest) -> AsyncIterator[Any]:
134+ # 1) 未 init 时直接失败并打 error
135+ if self._shutdown_done:
136+ logger.error("Access 已 shutdown,不再收消息")
137+ return
138+ if not self._response_parser:
139+ logger.error("ResponseParser 未设置")
140+ return
141+ if isinstance(msg, ISessionRequest):
142+ session_request = msg
143+ rid = session_request.request_id
144+ logger.debug(
145+ "Access receive session: session_id=%s session_conc=%s session_ttl=%s request_id=%s",
146+ session_request.session_id,
147+ session_request.session_concurrency,
148+ session_request.session_ttl,
149+ rid,
150+ )
151+ else:
152+ rid = getattr(msg, "request_id", None)
153+ if not rid:
154+ rid = uuid.uuid4().hex
155+ logger.info(
156+ "Access 自动生成 request_id=%s(请求未提供,多路复用必须非空)", rid
157+ )
158+ wd = getattr(msg, "wire_dict", None)
159+ if isinstance(wd, dict) and not wd.get("request_id"):
160+ # 对端按 request_id 回包路由;inplace 注入到原 wire_dict 即可
161+ wd["request_id"] = rid
162+ msg = _AutoIdRequest(msg, rid)
163+ logger.info("Access 收到请求: request_id=%s", rid)
164+ # 2) 策略层:从业务请求解析出 session_id、会话级并发、TTL
165+ if not self._strategy:
166+ logger.error("未设置session策略")
167+ return
168+ session_request = self._strategy.handle_session(msg)
169+ logger.debug(
170+ "Access 策略生成 session: session_id=%s session_conc=%s session_ttl=%s",
171+ session_request.session_id,
172+ session_request.session_concurrency,
173+ session_request.session_ttl,
174+ )
175+ 
176+ # 3) 每个入口请求独占一条响应队列 + cancel,用于多路复用/取消
177+ response_queue: asyncio.Queue[Any] = asyncio.Queue()
178+ cancel: asyncio.Future = asyncio.get_running_loop().create_future()
179+ wrapper = SessionRequestWrapper(session_request, response_queue, cancel)
180+ 
181+ # 4) 入用户队列,由 ServiceManager 异步消费并路由到具体服务实例
182+ await self._service_manager.handle_message(wrapper)
183+ logger.debug("Access 已将请求投递 ServiceManager, request_id=%s", rid)
184+ try:
185+ while True:
186+ try:
187+ to = self._config.message_timeout if self._config else 600
188+ data = await asyncio.wait_for(response_queue.get(), timeout=to)
189+ except asyncio.TimeoutError:
190+ # 等响应超时:结束迭代(业务上视为挂起/失败,由调用方处理)
191+ logger.error(
192+ "Access 等待下游响应超时: request_id=%s timeout=%s", rid, to
193+ )
194+ break
195+ logger.debug("Access 收到流式分片, request_id=%s", rid)
196+ yield self._response_parser.response(data)
197+ if self._response_parser.is_completed(data):
198+ logger.debug("Access 收到终态分片, request_id=%s", rid)
199+ break
200+ if cancel.done():
201+ logger.debug("Access 因 cancel 结束收包, request_id=%s", rid)
202+ break
203+ finally:
204+ if not cancel.done():
205+ cancel.set_result(None)
206+ logger.debug("Access send_message 协程结束, request_id=%s", rid)
@@ -0,0 +1,77 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+"""Docker 容器级部署:提供与 K8s 同构的 deploy/delete。生产可接 docker engine API;无依赖环境用 Mock。"""
5+ 
6+from __future__ import annotations
7+ 
8+import uuid
9+from dataclasses import dataclass
10+from typing import Any, Optional
11+ 
12+from openjiuwen_runtime.foundation.log import get_logger
13+ 
14+from .runtime import IDeployController
15+ 
16+logger = get_logger(__name__)
17+ 
18+ 
19+@dataclass(frozen=True)
20+class DockerRunInfo:
21+ """容器拉起后的可访问点。"""
22+ 
23+ container_id: str
24+ host: str
25+ port: int
26+ 
27+ 
28+class DockerServiceHandler:
29+ """占位实现:不强制依赖 docker 库;可替换为 aiodocker 等。"""
30+ 
31+ def __init__(
32+ self,
33+ image: str,
34+ *,
35+ host: str = "127.0.0.1",
36+ publish_port: int = 8000,
37+ ) -> None:
38+ if not image:
39+ raise ValueError("image is required")
40+ self._image = image
41+ self._host = host
42+ self._port = int(publish_port)
43+ self._container_id: Optional[str] = None
44+ logger.debug("DockerServiceHandler 初始化: image=%s host=%s port=%s", image, host, self._port)
45+ 
46+ @property
47+ def container_id(self) -> Optional[str]:
48+ return self._container_id
49+ 
50+ async def deploy(self) -> DockerRunInfo:
51+ # 生产可替换为 aiodocker run + 健康检查
52+ logger.debug("Docker deploy(桩) 开始: image=%s", self._image)
53+ self._container_id = f"ctr-{uuid.uuid4().hex[:12]}"
54+ info = DockerRunInfo(container_id=self._container_id, host=self._host, port=self._port)
55+ logger.info("Docker deploy(桩) 成功: %s", info)
56+ return info
57+ 
58+ async def delete(self) -> str:
59+ cid = self._container_id or "unknown"
60+ self._container_id = None
61+ logger.info("Docker delete(桩) 完成: container_id=%s", cid)
62+ return cid
63+ 
64+ 
65+class DockerDeployController:
66+ def __init__(self, inner: DockerServiceHandler) -> None:
67+ self._inner = inner
68+ 
69+ @property
70+ def resource_id(self) -> str | None:
71+ return self._inner.container_id
72+ 
73+ async def deploy(self) -> Any:
74+ return await self._inner.deploy()
75+ 
76+ async def delete(self) -> str:
77+ return await self._inner.delete()
@@ -0,0 +1,107 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+"""双 asyncio.Queue:系统(内部)消息优先于用户消息,均为可配置 maxsize。"""
5+ 
6+from __future__ import annotations
7+ 
8+import asyncio
9+from typing import Generic, TypeVar
10+ 
11+from openjiuwen_runtime.foundation.log import get_logger
12+ 
13+logger = get_logger(__name__)
14+ 
15+T = TypeVar("T")
16+ 
17+ 
18+class PriorityDualAsyncQueues(Generic[T]):
19+ """先 drain 系统队列,再等待双队列;系统消息始终优先。"""
20+ 
21+ def __init__(self, system_maxsize: int, user_maxsize: int) -> None:
22+ self._sys: asyncio.Queue[T] = asyncio.Queue(maxsize=system_maxsize)
23+ self._user: asyncio.Queue[T] = asyncio.Queue(maxsize=user_maxsize)
24+ self._closed = False
25+ logger.debug(
26+ "双队列已创建: system_maxsize=%s user_maxsize=%s", system_maxsize, user_maxsize
27+ )
28+ 
29+ @property
30+ def closed(self) -> bool:
31+ return self._closed
32+ 
33+ def mark_closed(self) -> None:
34+ self._closed = True
35+ logger.info("双队列已标记关闭, 不再接受入队")
36+ 
37+ @property
38+ def system_maxsize(self) -> int:
39+ return self._sys.maxsize # type: ignore[no-any-return, attr-defined]
40+ 
41+ @property
42+ def user_maxsize(self) -> int:
43+ return self._user.maxsize # type: ignore[no-any-return, attr-defined]
44+ 
45+ def system_qsize(self) -> int:
46+ return self._sys.qsize()
47+ 
48+ def user_qsize(self) -> int:
49+ return self._user.qsize()
50+ 
51+ async def put_user(self, item: T) -> None:
52+ if self._closed:
53+ logger.error("双队列已关闭, 拒绝 put_user")
54+ raise RuntimeError("PriorityDualAsyncQueues is closed")
55+ await self._user.put(item)
56+ logger.debug("用户队列入队, 当前~长度=%s", self._user.qsize())
57+ 
58+ async def put_system(self, item: T) -> None:
59+ if self._closed:
60+ logger.error("双队列已关闭, 拒绝 put_system")
61+ raise RuntimeError("PriorityDualAsyncQueues is closed")
62+ await self._sys.put(item)
63+ logger.debug("系统队列入队, 当前~长度=%s", self._sys.qsize())
64+ 
65+ async def get(self) -> T:
66+ """阻塞获取下一条消息;系统队列有数据时先返回系统侧。"""
67+ if self._closed and self._sys.empty() and self._user.empty():
68+ raise RuntimeError("PriorityDualAsyncQueues is closed and empty")
69+ while True:
70+ if self._closed and self._sys.empty() and self._user.empty():
71+ raise RuntimeError("PriorityDualAsyncQueues is closed and empty")
72+ while True:
73+ try:
74+ item = self._sys.get_nowait()
75+ logger.debug("双队列 get: 取系统项, sys~=%s user~=%s", self._sys.qsize(), self._user.qsize())
76+ return item
77+ except asyncio.QueueEmpty:
78+ break
79+ t_sys = asyncio.create_task(self._sys.get())
80+ t_usr = asyncio.create_task(self._user.get())
81+ try:
82+ done, pending = await asyncio.wait(
83+ {t_sys, t_usr},
84+ return_when=asyncio.FIRST_COMPLETED,
85+ )
86+ if len(done) != 1:
87+ raise RuntimeError("PriorityDualAsyncQueues is closed ")
88+ d = next(iter(done))
89+ for p in pending:
90+ p.cancel()
91+ try:
92+ await p
93+ except (asyncio.CancelledError, Exception):
94+ pass
95+ if d is t_sys:
96+ out = t_sys.result()
97+ logger.debug("双队列 get: 阻塞取系统项, user~=%s", self._user.qsize())
98+ else:
99+ out = t_usr.result()
100+ logger.debug("双队列 get: 阻塞取用户项, sys~=%s", self._sys.qsize())
101+ return out
102+ except Exception:
103+ for p in (t_sys, t_usr):
104+ if not p.done():
105+ p.cancel()
106+ logger.error("双队列 get 等待异常", exc_info=True)
107+ raise
@@ -0,0 +1,19 @@
1+from .interfaces import ResponseMessage
2+ 
3+ERROR_CODE_MAPPING = {
4+ -1: ("内部错误",
5+ "内部错误"),
6+ 100001: ("服务并发度超过上限,消息请求失败",
7+ "服务并发度超过上限,消息请求失败"),
8+ 100002: ("服务启动失败",
9+ "服务启动失败"),
10+}
11+ 
12+ 
13+def exception_message(code: int, language: str = "cn") -> ResponseMessage:
14+ if code not in ERROR_CODE_MAPPING:
15+ code = -1
16+ if language == "cn":
17+ return ResponseMessage(code, ERROR_CODE_MAPPING[code][0])
18+ else:
19+ return ResponseMessage(code, ERROR_CODE_MAPPING[code][1])
@@ -0,0 +1,323 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+import asyncio
5+from abc import ABC, abstractmethod
6+from typing import (
7+ Any,
8+ Awaitable,
9+ Callable,
10+ Optional,
11+ TYPE_CHECKING,
12+ AsyncIterator,
13+ Protocol,
14+ TypeAlias,
15+ runtime_checkable,
16+)
17+ 
18+from .models import MessagePriority, MessageType
19+ 
20+if TYPE_CHECKING:
21+ from .models import AccessConfig, SessionConfig
22+ 
23+# 与 ``ServiceHandler`` 传入的 ``async def on_request_complete(r)`` 一致(可 await)
24+OnRequestCompleteCallback: TypeAlias = Callable[[Optional[str]], Awaitable[None]]
25+ 
26+ 
27+class PriorityMessage(ABC):
28+ @property
29+ @abstractmethod
30+ def priority(self) -> "MessagePriority":
31+ pass
32+ 
33+ 
34+class RawMessage(PriorityMessage):
35+ def __init__(
36+ self,
37+ message_type: MessageType,
38+ message: Any,
39+ priority: MessagePriority = MessagePriority.LOW,
40+ ) -> None:
41+ self.message_type = message_type
42+ self.message = message
43+ self._priority = priority
44+ 
45+ @property
46+ def priority(self) -> "MessagePriority":
47+ return self._priority
48+ 
49+ 
50+class IRequest(ABC):
51+ @property
52+ @abstractmethod
53+ def request_id(self) -> Optional[str]:
54+ pass
55+ 
56+ @property
57+ @abstractmethod
58+ def chat_id(self) -> Optional[str]:
59+ pass
60+ 
61+ @property
62+ @abstractmethod
63+ def user_id(self) -> Optional[str]:
64+ pass
65+ 
66+ @property
67+ @abstractmethod
68+ def bot_id(self) -> Optional[str]:
69+ pass
70+ 
71+ @property
72+ @abstractmethod
73+ def session_id(self) -> Optional[str]:
74+ pass
75+ 
76+ 
77+class ISessionRequest(PriorityMessage):
78+ @property
79+ @abstractmethod
80+ def session_id(self) -> str:
81+ pass
82+ 
83+ @property
84+ @abstractmethod
85+ def session_concurrency(self) -> int:
86+ pass
87+ 
88+ @property
89+ @abstractmethod
90+ def session_ttl(self) -> int:
91+ pass
92+ 
93+ @property
94+ @abstractmethod
95+ def request_id(self) -> Optional[str]:
96+ pass
97+ 
98+ @property
99+ @abstractmethod
100+ def raw_msg(self) -> Any:
101+ pass
102+ 
103+ 
104+class ResponseMessage:
105+ def __init__(self, code: int, message: str, data: dict = None) -> None:
106+ self.code = code
107+ self.message = message
108+ self.data = data
109+ 
110+ 
111+class SessionRequestWrapper:
112+ def __init__(
113+ self,
114+ request: ISessionRequest,
115+ response_queue: asyncio.Queue[Any],
116+ cancel: asyncio.Future[Any],
117+ ) -> None:
118+ self._session_request = request
119+ self._response_queue = response_queue
120+ self._cancel = cancel
121+ 
122+ @property
123+ def session_request(self) -> ISessionRequest:
124+ return self._session_request
125+ 
126+ @property
127+ def response_queue(self) -> asyncio.Queue[Any]:
128+ return self._response_queue
129+ 
130+ @property
131+ def cancel(self) -> asyncio.Future[Any]:
132+ return self._cancel
133+ 
134+ 
135+class IResponseParser(ABC):
136+ @abstractmethod
137+ def request_id(self, data: dict[str, Any]) -> Optional[str]:
138+ pass
139+ 
140+ @abstractmethod
141+ def is_completed(self, data: dict[str, Any]) -> bool:
142+ pass
143+ 
144+ @abstractmethod
145+ def response(self, data: dict[str, Any]) -> Any:
146+ pass
147+ 
148+ 
149+class ITimer(ABC):
150+ @abstractmethod
151+ async def start_timer(self, key: str, ttl: int, callback) -> None:
152+ pass
153+ 
154+ @abstractmethod
155+ async def cancel_timer(self, key: str) -> bool:
156+ pass
157+ 
158+ async def stop_all(self) -> None:
159+ """若实现支持,取消所有活动计时;默认无操作。"""
160+ return
161+ 
162+ 
163+class ISessionStrategy(ABC):
164+ @abstractmethod
165+ def handle_session(self, msg: IRequest) -> ISessionRequest:
166+ pass
167+ 
168+ @abstractmethod
169+ def configure(self, concurrency: int, ttl: int) -> None:
170+ """配置会话的并发度和 TTL"""
171+ pass
172+ 
173+ 
174+class IAccess(ABC):
175+ @abstractmethod
176+ async def init(
177+ self,
178+ response_parser: IResponseParser,
179+ strategy: ISessionStrategy,
180+ config: "AccessConfig",
181+ session_config: "SessionConfig",
182+ ) -> None:
183+ pass
184+ 
185+ @abstractmethod
186+ def send_message(self, msg: IRequest) -> AsyncIterator[Any]:
187+ pass
188+ 
189+ @abstractmethod
190+ async def shutdown(self) -> None:
191+ """优雅退出:停后台 task/双队列/计时器,并释放已拉起的各服务实例(如 Pod/连接)。"""
192+ pass
193+ 
194+ @abstractmethod
195+ async def update_config(
196+ self, config: "AccessConfig", session_config: Optional["SessionConfig"] = None
197+ ) -> None:
198+ """运行时热更新配置。存量 session/service 不变,新建的使用新值。"""
199+ pass
200+ 
201+ 
202+@runtime_checkable
203+class IServiceMessageChannel(Protocol):
204+ """与下游服务通信;单服务实例上通常是 **一条长连接**(如 WebSocket),上有多路并发流式 ``request_id``。
205+ 
206+ **实现方式**:用结构子类型实现本 Protocol(**不要**让具体子类再继承本 Protocol,以免与
207+ 类型检查器对 ``Protocol`` 子类化的限制冲突。)
208+ 
209+ 约定:
210+ * ``send`` 中 **上行** 发送一帧业务负载(通常 JSON 序列化自 ``ISessionRequest.raw_msg``);
211+ * **下行** 由实现类在独接收循环里按 ``IResponseParser.request_id`` 分片写入对应 ``SessionRequestWrapper.response_queue``;
212+ * 当某 ``request_id`` 的响应用 ``IResponseParser.is_completed`` 判定结束时,**必须**
213+ ``await on_request_complete(request_id)`` 归还本实例并发。
214+ 可选实现(鸭子类型): ``bind_handler(handler, parser)``、``on_pod_ready(service_id, pod_info)``、``close()``.
215+ """
216+ 
217+ async def send(self, service_id: str, wrapper: SessionRequestWrapper, *, response_parser: IResponseParser,
218+ on_request_complete: OnRequestCompleteCallback,
219+ ) -> None:
220+ pass
221+ 
222+ 
223+class IServiceInstanceFactory(ABC):
224+ @abstractmethod
225+ async def new_service(self, response_parser: IResponseParser) -> "IServiceHandler":
226+ pass
227+ 
228+ 
229+class IServiceManager(ABC):
230+ @abstractmethod
231+ async def init(self, response_parser: IResponseParser) -> None:
232+ pass
233+ 
234+ @abstractmethod
235+ async def start(self) -> None:
236+ pass
237+ 
238+ @abstractmethod
239+ async def stop(self) -> None:
240+ pass
241+ 
242+ @abstractmethod
243+ async def handle_message(self, msg: "SessionRequestWrapper") -> None:
244+ pass
245+ 
246+ @abstractmethod
247+ async def enqueue_system(self, event: Any) -> None:
248+ """投递内部高优先级消息(如缩容、运维事件)。"""
249+ 
250+ @abstractmethod
251+ async def update_config(self, **kwargs) -> None:
252+ """运行时更新调度参数并递增代际。"""
253+ 
254+ 
255+class IServiceHandler(ABC):
256+ @property
257+ @abstractmethod
258+ def id(self) -> str:
259+ pass
260+ 
261+ @property
262+ @abstractmethod
263+ def total_concurrency(self) -> int:
264+ pass
265+ 
266+ @property
267+ @abstractmethod
268+ def available_concurrency(self) -> int:
269+ pass
270+ 
271+ @abstractmethod
272+ def try_reserve_session_quota(self, session_id: str, quota: int) -> bool:
273+ """为新 session 预留 ``quota`` 点服务并发;同一 session 重复调用须幂等成功。"""
274+ 
275+ @property
276+ @abstractmethod
277+ def inflight_requests(self) -> int:
278+ """当前实例上通道在途请求数(消息粒度),用于 TTL / idle;不等同于服务额度占用。"""
279+ 
280+ @property
281+ @abstractmethod
282+ def active_session_count(self) -> int:
283+ pass
284+ 
285+ def open_session_ids(self) -> list[str]:
286+ """当前实例上仍占位的 session_id 列表;默认无。转入 idle 前可据此清理亲和路由。"""
287+ return []
288+ 
289+ def session_active_request_count(self, session_id: str) -> int:
290+ """指定 session 上仍在飞的请求数;默认 0。session_ttl 到期时用以判断是否可立即移除。"""
291+ return 0
292+ 
293+ @abstractmethod
294+ def has_session(self, session_id: str) -> bool:
295+ pass
296+ 
297+ @abstractmethod
298+ async def handle_message(self, msg: "SessionRequestWrapper") -> None:
299+ pass
300+ 
301+ @abstractmethod
302+ async def remove_session(self, session_id: str) -> int:
303+ pass
304+ 
305+ @abstractmethod
306+ async def deploy(self) -> None:
307+ pass
308+ 
309+ @abstractmethod
310+ async def delete(self) -> None:
311+ pass
312+ 
313+ def set_idle_pool_transition_hook(
314+ self, hook: Optional[Callable[[str], Awaitable[None]]]
315+ ) -> None:
316+ """默认无实现;`ServiceHandler` 会注入,供无业务时按 `service_ttl` 转入 idle 池。"""
317+ return
318+ 
319+ 
320+class ISessionHandler(ABC):
321+ @abstractmethod
322+ async def handle_message(self, msg: "SessionRequestWrapper") -> None:
323+ pass
@@ -0,0 +1,15 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+"""走系统(高优先级)队列的内部事件。"""
5+ 
6+from __future__ import annotations
7+ 
8+from dataclasses import dataclass
9+ 
10+ 
11+@dataclass(frozen=True)
12+class ServiceReclaimEvent:
13+ """空闲实例超过 service_ttl 后触发的缩容。"""
14+ 
15+ service_id: str
@@ -0,0 +1,381 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+"""K8s Pod 部署(kubernetes_asyncio),与业务层 ServiceHandler 解耦。生产环境用于 deploy/delete,测试可 Mock。"""
4+ 
5+from __future__ import annotations
6+ 
7+import asyncio
8+import re
9+import secrets
10+import string
11+from dataclasses import dataclass
12+from typing import Dict, Optional, Tuple
13+ 
14+from kubernetes_asyncio import client, config
15+from kubernetes_asyncio.client.rest import ApiException
16+ 
17+from openjiuwen_runtime.foundation.log import get_logger
18+ 
19+logger = get_logger(__name__)
20+ 
21+ 
22+@dataclass(frozen=True)
23+class PodDeployInfo:
24+ """Pod 部署成功后的信息,字段对应 `kubectl get pods -o wide` 的各列。"""
25+ 
26+ pod_name: str
27+ namespace: str
28+ port: int
29+ pod_ip: str
30+ host_ip: Optional[str] = None
31+ node_name: Optional[str] = None
32+ 
33+ 
34+class K8sServiceHandler:
35+ """仅负责 Pod 创建/删除。业务侧可组合到 ServiceHandler 的 deploy/delete。"""
36+ 
37+ _MAX_PREFIX_LEN = 47
38+ _NAME_INVALID_CHARS = re.compile(r"[^a-z0-9-]+")
39+ 
40+ def __init__(
41+ self,
42+ image: str,
43+ *,
44+ name_prefix: str = "jiuwenclaw",
45+ namespace: str = "default",
46+ pod_name: str = None,
47+ container_name: str = "jiuwenclaw-agentserver",
48+ container_port: int = 18092,
49+ port_name: str = "http1",
50+ image_pull_policy: str = "IfNotPresent",
51+ env_vars: Optional[Dict[str, str]] = None,
52+ extra_labels: Optional[Dict[str, str]] = None,
53+ restart_policy: str = "Always",
54+ readiness_initial_delay: int = 5,
55+ readiness_period: int = 10,
56+ kubeconfig: Optional[str] = None,
57+ ready_timeout: float = 300.0,
58+ ready_poll_interval: float = 2.0,
59+ delete_grace_period: int = 30,
60+ delete_timeout: float = 120.0,
61+ delete_poll_interval: float = 1.0,
62+ nfs_server: Optional[str] = None, # NFS 服务器地址
63+ nfs_path: Optional[str] = None, # NFS 共享路径
64+ nfs_mount_path: Optional[str] = None, # 容器内挂载路径
65+ cpu_request: str = "500m",
66+ memory_request: str = "1Gi",
67+ cpu_limit: Optional[str] = None,
68+ memory_limit: Optional[str] = None,
69+ ):
70+ if not image:
71+ raise ValueError("image is required")
72+ 
73+ self._image = image
74+ self._name_prefix = pod_name if pod_name else self._sanitize_prefix(name_prefix)
75+ self._namespace = namespace
76+ self._container_name = container_name
77+ self._container_port = int(container_port)
78+ self._port_name = port_name
79+ self._image_pull_policy = image_pull_policy
80+ self._env_vars: Dict[str, str] = dict(env_vars or {})
81+ self._extra_labels: Dict[str, str] = dict(extra_labels or {})
82+ self._restart_policy = restart_policy
83+ self._readiness_initial_delay = int(readiness_initial_delay)
84+ self._readiness_period = int(readiness_period)
85+ self._kubeconfig = kubeconfig
86+ self._ready_timeout = float(ready_timeout)
87+ self._ready_poll_interval = float(ready_poll_interval)
88+ self._delete_grace_period = int(delete_grace_period)
89+ self._delete_timeout = float(delete_timeout)
90+ self._delete_poll_interval = float(delete_poll_interval)
91+ self._nfs_server = nfs_server
92+ self._nfs_path = nfs_path
93+ self._nfs_mount_path = nfs_mount_path
94+ 
95+ self._cpu_request = cpu_request
96+ self._memory_request = memory_request
97+ self._cpu_limit = cpu_limit if cpu_limit is not None else cpu_request
98+ self._memory_limit = memory_limit if memory_limit is not None else memory_request
99+ 
100+ self._pod_name: Optional[str] = None
101+ self._config_loaded = False
102+ 
103+ @property
104+ def pod_name(self) -> Optional[str]:
105+ return self._pod_name
106+ 
107+ @classmethod
108+ def _sanitize_prefix(cls, prefix: str) -> str:
109+ if not prefix:
110+ raise ValueError("name_prefix must not be empty")
111+ cleaned = cls._NAME_INVALID_CHARS.sub("-", prefix.lower()).strip("-")
112+ if not cleaned:
113+ raise ValueError(f"name_prefix {prefix!r} contains no valid DNS-1123 chars")
114+ return cleaned[: cls._MAX_PREFIX_LEN]
115+ 
116+ @staticmethod
117+ def _random_suffix(length: int) -> str:
118+ alphabet = string.ascii_lowercase + string.digits
119+ return "".join(secrets.choice(alphabet) for _ in range(length))
120+ 
121+ def _generate_pod_name(self) -> str:
122+ return f"{self._name_prefix}-{self._random_suffix(10)}-{self._random_suffix(5)}"
123+ 
124+ async def _ensure_config(self) -> None:
125+ if self._config_loaded:
126+ return
127+ try:
128+ config.load_incluster_config()
129+ logger.debug("K8s 已加载 in-cluster 配置")
130+ except config.ConfigException:
131+ if self._kubeconfig:
132+ await config.load_kube_config(config_file=self._kubeconfig)
133+ logger.debug("K8s 已加载 kubeconfig: %s", self._kubeconfig)
134+ else:
135+ await config.load_kube_config()
136+ logger.debug("K8s 已加载默认 kubeconfig")
137+ self._config_loaded = True
138+ 
139+ def _build_pod_body(self, pod_name: str) -> client.V1Pod:
140+ env_list = [client.V1EnvVar(name=k, value=str(v)) for k, v in self._env_vars.items()]
141+ labels: Dict[str, str] = {"app": pod_name}
142+ if self._extra_labels:
143+ labels.update(self._extra_labels)
144+ 
145+ volume_mounts = []
146+ volumes = []
147+ 
148+ if self._nfs_server and self._nfs_path and self._nfs_mount_path:
149+ nfs_volume_name = "nfs-volume"
150+ volumes.append(
151+ client.V1Volume(
152+ name=nfs_volume_name,
153+ nfs=client.V1NFSVolumeSource(
154+ server=self._nfs_server,
155+ path=self._nfs_path,
156+ ),
157+ )
158+ )
159+ volume_mounts.append(
160+ client.V1VolumeMount(
161+ name=nfs_volume_name,
162+ mount_path=self._nfs_mount_path,
163+ )
164+ )
165+ 
166+ container_resources = client.V1ResourceRequirements(
167+ requests={
168+ "cpu": self._cpu_request,
169+ "memory": self._memory_request,
170+ },
171+ limits={
172+ "cpu": self._cpu_limit,
173+ "memory": self._memory_limit,
174+ },
175+ )
176+ 
177+ container = client.V1Container(
178+ name=self._container_name,
179+ image=self._image,
180+ image_pull_policy=self._image_pull_policy,
181+ ports=[client.V1ContainerPort(name=self._port_name, container_port=self._container_port)],
182+ env=env_list or None,
183+ volume_mounts=volume_mounts or None,
184+ resources=container_resources,
185+ readiness_probe=client.V1Probe(
186+ tcp_socket=client.V1TCPSocketAction(port=self._container_port),
187+ initial_delay_seconds=self._readiness_initial_delay,
188+ period_seconds=self._readiness_period,
189+ ),
190+ )
191+ 
192+ return client.V1Pod(
193+ api_version="v1",
194+ kind="Pod",
195+ metadata=client.V1ObjectMeta(
196+ name=pod_name, namespace=self._namespace, labels=labels
197+ ),
198+ spec=client.V1PodSpec(
199+ containers=[container],
200+ restart_policy=self._restart_policy,
201+ volumes=volumes or None,
202+ ),
203+ )
204+ 
205+ async def deploy(self) -> PodDeployInfo:
206+ # 1) 加载集群访问配置 2) 创建 Pod 3) 轮询至 Running + Ready + 有 podIP
207+ await self._ensure_config()
208+ 
209+ pod_name = self._generate_pod_name()
210+ body = self._build_pod_body(pod_name)
211+ logger.info(
212+ "K8s 创建 Pod: name=%s namespace=%s image=%s", pod_name, self._namespace, self._image
213+ )
214+ logger.debug(
215+ "Pod 规约: container=%s port=%s resources(cpu/mem)=requests[%s/%s] limits[%s/%s]",
216+ self._container_name,
217+ self._container_port,
218+ self._cpu_request,
219+ self._memory_request,
220+ self._cpu_limit,
221+ self._memory_limit,
222+ )
223+ 
224+ api_client = client.ApiClient()
225+ try:
226+ core = client.CoreV1Api(api_client)
227+ 
228+ try:
229+ await core.create_namespaced_pod(namespace=self._namespace, body=body)
230+ except ApiException as exc:
231+ if exc.status == 409:
232+ retry_name = self._generate_pod_name()
233+ logger.warning("Pod name conflict %s, retrying with %s", pod_name, retry_name)
234+ body.metadata.name = retry_name
235+ if body.metadata.labels is None:
236+ body.metadata.labels = {}
237+ body.metadata.labels["app"] = retry_name
238+ await core.create_namespaced_pod(namespace=self._namespace, body=body)
239+ pod_name = retry_name
240+ else:
241+ raise
242+ 
243+ self._pod_name = pod_name
244+ pod_ip, host_ip, node_name = await self._wait_running_ready(core, pod_name)
245+ logger.info(
246+ "Pod ready: name=%s, pod_ip=%s, node=%s",
247+ pod_name,
248+ pod_ip,
249+ node_name,
250+ )
251+ return PodDeployInfo(
252+ pod_name=pod_name,
253+ namespace=self._namespace,
254+ port=self._container_port,
255+ pod_ip=pod_ip,
256+ host_ip=host_ip,
257+ node_name=node_name,
258+ )
259+ finally:
260+ await api_client.close()
261+ 
262+ async def _wait_running_ready(
263+ self, core: client.CoreV1Api, pod_name: str
264+ ) -> Tuple[str, Optional[str], Optional[str]]:
265+ loop = asyncio.get_running_loop()
266+ deadline = loop.time() + self._ready_timeout
267+ last_reason = ""
268+ 
269+ while True:
270+ pod = await core.read_namespaced_pod(name=pod_name, namespace=self._namespace)
271+ status = pod.status
272+ phase = (status.phase or "") if status else ""
273+ 
274+ if phase in ("Failed", "Succeeded"):
275+ raise RuntimeError(
276+ f"Pod {pod_name} entered terminal phase {phase}: {last_reason}"
277+ )
278+ 
279+ container_statuses = (status.container_statuses or []) if status else []
280+ all_containers_ready = bool(container_statuses) and all(
281+ bool(cs.ready) for cs in container_statuses
282+ )
283+ ready_cond_true = bool(status) and any(
284+ c.type == "Ready" and c.status == "True" for c in (status.conditions or [])
285+ )
286+ pod_ip = (status.pod_ip if status else None) or ""
287+ 
288+ is_running_and_containers_ready = phase == "Running" and all_containers_ready
289+ is_ready_with_ip = ready_cond_true and pod_ip
290+ if is_running_and_containers_ready and is_ready_with_ip:
291+ host_ip = status.host_ip if status else None
292+ node_name = getattr(pod.spec, "node_name", None) if pod.spec else None
293+ return pod_ip, host_ip, node_name
294+ 
295+ for cs in container_statuses:
296+ waiting = getattr(cs.state, "waiting", None) if cs.state else None
297+ reason = getattr(waiting, "reason", None) if waiting else None
298+ if reason:
299+ last_reason = (
300+ f"container={cs.name} waiting={reason} "
301+ f"msg={getattr(waiting, 'message', '') or ''}"
302+ )
303+ 
304+ if loop.time() >= deadline:
305+ logger.error(
306+ "K8s Pod 就绪超时: name=%s phase=%s last_reason=%s", pod_name, phase, last_reason
307+ )
308+ raise TimeoutError(
309+ f"Pod {pod_name} not Running 1/1 within {self._ready_timeout}s "
310+ f"(phase={phase!r}, last_reason={last_reason!r})"
311+ )
312+ logger.debug("K8s 等待 Pod 就绪: name=%s phase=%s", pod_name, phase)
313+ await asyncio.sleep(self._ready_poll_interval)
314+ 
315+ async def delete(self) -> str:
316+ if not self._pod_name:
317+ raise RuntimeError("delete() called before deploy() or pod already deleted")
318+ 
319+ pod_name = self._pod_name
320+ await self._ensure_config()
321+ logger.info("Deleting pod: name=%s, namespace=%s", pod_name, self._namespace)
322+ 
323+ api_client = client.ApiClient()
324+ try:
325+ core = client.CoreV1Api(api_client)
326+ try:
327+ await core.delete_namespaced_pod(
328+ name=pod_name,
329+ namespace=self._namespace,
330+ body=client.V1DeleteOptions(
331+ grace_period_seconds=self._delete_grace_period,
332+ propagation_policy="Foreground",
333+ ),
334+ )
335+ except ApiException as exc:
336+ if exc.status != 404:
337+ raise
338+ logger.info("Pod %s already absent; treating delete as idempotent", pod_name)
339+ await self._wait_pod_deleted(core, pod_name)
340+ finally:
341+ await api_client.close()
342+ 
343+ logger.info("Pod deleted: name=%s", pod_name)
344+ self._pod_name = None
345+ return pod_name
346+ 
347+ async def _wait_pod_deleted(self, core: client.CoreV1Api, pod_name: str) -> None:
348+ loop = asyncio.get_running_loop()
349+ deadline = loop.time() + self._delete_timeout
350+ 
351+ while True:
352+ try:
353+ await core.read_namespaced_pod(name=pod_name, namespace=self._namespace)
354+ except ApiException as exc:
355+ if exc.status == 404:
356+ return
357+ raise
358+ 
359+ if loop.time() >= deadline:
360+ logger.error("K8s Pod 删除超时: name=%s", pod_name)
361+ raise TimeoutError(
362+ f"Pod {pod_name} not deleted within {self._delete_timeout}s"
363+ )
364+ await asyncio.sleep(self._delete_poll_interval)
365+ 
366+ 
367+class K8sDeployController:
368+ """将 K8sServiceHandler 适配为 session.runtime.IDeployController。"""
369+ 
370+ def __init__(self, k8s: K8sServiceHandler) -> None:
371+ self._k8s = k8s
372+ 
373+ @property
374+ def resource_id(self) -> Optional[str]:
375+ return self._k8s.pod_name
376+ 
377+ async def deploy(self) -> PodDeployInfo:
378+ return await self._k8s.deploy()
379+ 
380+ async def delete(self) -> str:
381+ return await self._k8s.delete()
@@ -0,0 +1,70 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+"""Session SDK 数据模型"""
5+ 
6+from __future__ import annotations
7+ 
8+from dataclasses import dataclass
9+from enum import Enum
10+from typing import TYPE_CHECKING, Optional
11+ 
12+if TYPE_CHECKING:
13+ from openjiuwen_runtime.foundation.db import DBHandler
14+ 
15+ 
16+class MessagePriority(str, Enum):
17+ """消息优先级"""
18+ 
19+ LOW = "low"
20+ MEDIUM = "medium"
21+ HIGH = "high"
22+ 
23+ 
24+class MessageType(str, Enum):
25+ USER_REQUEST = "user_request"
26+ 
27+ 
28+@dataclass
29+class SessionConfig:
30+ """单 session 维度:与 Access 中策略一起写入 ISessionRequest。"""
31+ 
32+ concurrency: int = 1
33+ """同一 session 内最大并行处理中的请求数(会话级限流)。"""
34+ ttl: int = 300
35+ """session 亲和保活窗口 (秒): 请求结束起计时, 期间收到该 session 的新消息会**取消**计时;
36+ 超时则移除 session handler、归还其在 service_handler 上占用的并发位。
37+ 设为 ``0`` 表示请求一结束、且该 session 无 inflight 时**立刻**释放, 不保留亲和。"""
38+ 
39+ 
40+@dataclass
41+class AccessConfig:
42+ """入口与 ServiceManager 的共享运行配置。"""
43+ 
44+ # 双队列
45+ user_queue_size: int = 1000
46+ system_queue_size: int = 100
47+ 
48+ # 业务与部署
49+ image: str = "app:latest"
50+ db_handler: Optional["DBHandler"] = None
51+ service_concurrency: int = 200
52+ min_idle_services: int = 1
53+ """池中**至少**保持的**空闲**实例数(仅由 autoscale 补位,**不会**因 ``service_ttl`` 从 idle 删 Pod)。
54+ 若 ``len(_idle) < min_idle``,约每 ``autoscale_interval`` 会 deploy 预热实例入 idle。设为 ``0`` 表示不维护热备。"""
55+ max_services: int = 10
56+ target_port: int = 8000
57+ invoke_path: str = "" # 与 pod_ip:target_port 组成 WS URI;空则仅 ws://host:port
58+ ws_use_tls: bool = False # True 为 wss://,否为 ws://
59+ 
60+ service_ttl: int = 300
61+ """① **in_use** 实例在「所有 session 均已归还 + 无 inflight」后, 等待本秒数再**转入** ``_idle``;
62+ 等待期间若又分配到新消息则取消等待, 实例继续占用。
63+ ② 当 **len(idle) > min_idle_services** 时, 多余 idle 立即被回收 (删 Pod);
64+ 若多余实例是**刚从 in_use 转入** idle 的, 由于 in_use 阶段已等过一次本字段, 不再叠加二次等待。
65+ 底数 min 台 idle 不会被本字段删除。设为 ``0`` 时①阶段无等待。负值行为未定义, 请勿使用。
66+ """
67+ 
68+ message_timeout: int = 600
69+ max_retries: int = 3
70+ autoscale_interval: float = 0.2
@@ -0,0 +1,74 @@
1+# coding: utf-8
2+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
3+ 
4+"""Session 映射管理"""
5+ 
6+import asyncio
7+from typing import Optional, Dict
8+ 
9+ 
10+class SessionRouter:
11+ """Session 映射管理,支持并发安全访问"""
12+ 
13+ def __init__(self) -> None:
14+ self._lock = asyncio.Lock()
15+ self._request_session: Dict[str, str] = {}
16+ 
17+ async def get_request_session(self, request_id: str) -> Optional[str]:
18+ """获取 request_id 对应的 session_id"""
19+ async with self._lock:
20+ return self._request_session.get(request_id)
21+ 
22+ async def set_request_session(self, request_id: str, session_id: str) -> None:
23+ """设置 request_id 到 session_id 的映射"""
24+ async with self._lock:
25+ self._request_session[request_id] = session_id
26+ 
27+ async def delete_request_session(self, request_id: str) -> bool:
28+ """删除 request_id 的映射"""
29+ async with self._lock:
30+ if request_id in self._request_session:
31+ del self._request_session[request_id]
32+ return True
33+ return False
34+ 
35+ async def clear(self) -> None:
36+ """清空 request → session 映射(服务缩容/销毁时调用)。"""
37+ async with self._lock:
38+ self._request_session.clear()
39+ 
40+ async def get_request_session_size(self) -> int:
41+ """获取 request_session 映射大小"""
42+ async with self._lock:
43+ return len(self._request_session)
44+ 
45+ 
46+class ServiceRouter:
47+ """Service 映射管理,支持并发安全访问"""
48+ 
49+ def __init__(self) -> None:
50+ self._lock = asyncio.Lock()
51+ self._session_service: Dict[str, str] = {}
52+ 
53+ async def get_session_service(self, session_id: str) -> Optional[str]:
54+ """获取 session_id 对应的 service_id"""
55+ async with self._lock:
56+ return self._session_service.get(session_id)
57+ 
58+ async def set_session_service(self, session_id: str, service_id: str) -> None:
59+ """设置 session_id 到 service_id 的映射"""
60+ async with self._lock:
61+ self._session_service[session_id] = service_id
62+ 
63+ async def delete_session_service(self, session_id: str) -> bool:
64+ """删除 session_id 的映射"""
65+ async with self._lock:
66+ if session_id in self._session_service:
67+ del self._session_service[session_id]
68+ return True
69+ return False
70+ 
71+ async def clear(self) -> None:
72+ """清空 session_id → service_id 的亲和表(如进程退出/优雅停机)。"""
73+ async with self._lock:
74+ self._session_service.clear()