尧图网站建设 
SkyPilot 管理策略服务实战用 FastAPI 构建 RESTful Admin Policy Server【免费下载链接】skypilotThe AI Compute Platform for frontier teams. SkyPilot turns fragmented AI compute into one AI supercomputer, so frontier AI teams build custom intelligence faster.项目地址: https://gitcode.com/GitHub_Trending/sk/skypilotSkyPilot 的 Admin Policy管理策略机制允许平台管理员对所有用户的sky launch/sky exec等请求执行自定义的校验与改写逻辑。本指南以仓库 examples/admin_policy/example_server 为例完整讲解如何用 FastAPI 将一个或多个 Admin Policy 封装成独立的 RESTful 策略服务并通过admin_policy: http://localhost:8080接入 SkyPilot。读完本文你将掌握 RESTful 策略服务的完整生命周期理解UserRequest/MutatedUserRequest的编解码约定、读懂示例服务的核心实现、学会动态加载外部策略包并能为自己的组织定制可独立部署的策略服务。一、为什么需要 RESTful 形式的 Admin PolicySkyPilot 采用客户端—服务端架构一个集中式的 API Server 可以统一承载用户的请求管理员则通过 Admin Policy 在请求进入核心流程之前注入校验 改写逻辑。在 docs/source/cloud-setup/policy.rst 中官方将部署方式划分为两类Python 包方式把实现sky.AdminPolicy的类发布为 Python 包安装到 SkyPilot 所在的 Python 环境并在配置中指定admin_policy: mypackage.MyPolicyRESTful 服务方式把策略逻辑托管在一个独立 HTTP 服务中配置中直接填写 URL例如admin_policy: https://example.com/policy。RESTful 方式的核心优势在于策略逻辑与 SkyPilot 环境完全解耦策略可以用任意语言实现只要遵循 API 约定可以集中部署、热更新且天然适合多用户、多客户端的场景——所有用户请求都汇聚到同一个策略端点统一裁决。代价是策略运行在服务端所在的主机无法访问发起请求的用户的本地环境如本地文件、本地凭证这一点官方文档也明确提示参见 policy.rst 中的说明。本文涉及的example_server目录正是这种 RESTful 方式的官方参考实现examples/admin_policy/example_server/ ├── README.md # 本指南对应的原始文档 ├── policy_server.py # 基础版内置策略组合 FastAPI 服务 └── dynamic_policy_server.py# 进阶版支持 --policy 动态加载外部策略二、准备工作与依赖2.1 安装 SkyPilot策略服务的核心数据类型sky.UserRequest、sky.MutatedUserRequest、sky.AdminPolicy都来自 SkyPilot 本身因此必须先安装 SkyPilotpip install skypilot安装完成后import sky即可获得策略接口与请求对象接口定义见 sky/admin_policy.py。2.2 安装示例策略包示例服务会引用example_policy包中预置的策略类如UseSpotForGpuPolicy。该包位于仓库 examples/admin_policy/example_policy安装方式如下在克隆仓库后执行cd skypilot pip install examples/admin_policy/example_policy该包通过 pyproject.toml 声明name example_policy、version 0.0.1并在init.py 中导出全部预置策略包括DoNothingPolicy、UseSpotForGpuPolicy、RejectAllPolicy、AddLabelsPolicy、DisablePublicIpPolicy、EnforceAutostopPolicy、RateLimitLaunchPolicy、GPUStaticQuotaPolicy、RejectOldClientsPolicy等。2.3 服务端运行时依赖示例服务基于 FastAPI 与 uvicorn 编写启动前需确保已安装pip install fastapi uvicorn三、启动策略服务按 README 的 Quickstart启动命令为python policy_server.py --policy DoNothingPolicy服务默认监听0.0.0.0:8080。这里需要说明一个仓库内的实现细节当前仓库中真正支持--policy参数的是 dynamic_policy_server.py而 policy_server.py 的 argparse 只暴露了--host与--port策略组合直接硬编码在policies列表中。两个文件均可独立启动服务区别见下文第五、六节脚本策略来源额外端点policy_server.py硬编码组合DoNothingPolicyUseSpotForGpuPolicyPOST /dynamic_policy_server.py--policy动态加载example_policy中的任意策略GET /健康检查、POST /set_autostop四、把策略服务接入 SkyPilot服务启动后在 SkyPilot 配置中设置admin_policy字段指向该服务admin_policy: http://localhost:8080仓库 examples/admin_policy/restful_policy.yaml 提供的正是这一最小配置。之后执行sky launch时SkyPilot 会先调用该 URL 对请求做校验与改写再继续后续流程。4.1 SkyPilot 如何识别并调用 URL 策略从源码看URL 与 Python 包路径的区分发生在 sky/utils/admin_policy_utils.py 的_get_policy_impl中_is_url()通过urlparse判断policy_string的 scheme 是否为http/https若为 URL则构造内置的RestfulAdminPolicy(policy_location)否则按module_path.class_name的格式importlib动态导入策略类。4.2 RestfulAdminPolicy 的调用约定RESTful 调用端的实现RestfulAdminPolicy完整定义了客户端与服务端的契约向policy_url发起POST请求body 为user_request.encode()的 JSON 字符串Content-Type: application/json默认超时 30 秒收到 HTTP 400 时抛出UserRequestRejectedByPolicy即策略主动拒绝请求其他非 2xx 状态码触发response.raise_for_status()网络类异常统一包装为RestfulPolicyError成功时用MutatedUserRequest.decode(response.json(), user_request)反序列化改写后的请求并恢复内部字段managed_job_dag、service_name。这意味策略服务的响应必须符合MutatedUserRequest的 JSON 格式否则会报Failed to decode response from admin policy URL。五、基础版服务剖析policy_server.pypolicy_server.py 是 README 指向的核心实现完整展示了请求进来 → 解码 → 逐策略应用 → 编码返回的完整链路。5.1 在服务内实现一个策略DoNothingPolicy是最简单的sky.AdminPolicy子类——不校验、不改写原样返回任务与配置class DoNothingPolicy(sky.AdminPolicy): Example policy: do nothing. classmethod def validate_and_mutate( cls, user_request: sky.UserRequest) - sky.MutatedUserRequest: Returns the user request unchanged. return sky.MutatedUserRequest(user_request.task, user_request.skypilot_config)这正是 AdminPolicy 抽象接口 要求的唯一抽象方法签名validate_and_mutate(cls, user_request) - MutatedUserRequest。需要拒绝请求时策略内raise异常即可服务端会统一转为 HTTP 400。5.2 唯一的策略端点POST /apply_policy处理器是服务的心脏app.post(/) async def apply_policy(request: Request) - JSONResponse: Apply an admin policy loaded from external package to a user request # Decode json_data await request.json() user_request sky.UserRequest.decode(json_data) # Example: change the following list to apply different policies. policies: List[sky.AdminPolicy] [ # Example: policy that implemented in the server package. DoNothingPolicy, # Example: policy from third party packages. example_policy.UseSpotForGpuPolicy, ] try: for policy in policies: mutated_request policy.validate_and_mutate(user_request) user_request.task mutated_request.task user_request.skypilot_config mutated_request.skypilot_config except Exception as e: # pylint: disablebroad-except return JSONResponse(contentstr(e), status_code400) return JSONResponse(contentmutated_request.encode())要点拆解解码sky.UserRequest.decode(json_data)将请求 body 还原为UserRequest内部通过 pydantic 模型_UserRequestBody校验再把 YAML 字符串反序列化为sky.Task与sky.Config见 sky/admin_policy.py组合策略policies列表是策略管道pipeline按顺序逐个执行前一个策略改写后的task/skypilot_config会覆盖回user_request供下一个策略继续处理。这里展示了两种来源的混用服务内实现的DoNothingPolicy 外部包example_policy提供的UseSpotForGpuPolicy拒绝语义任一策略抛异常整个请求被拒绝响应 400 并携带异常文本——与 SkyPilot 客户端 400 →UserRequestRejectedByPolicy的处理完全对应编码mutated_request.encode()输出符合_MutatedUserRequestBody结构的 JSONtaskskypilot_config两个 YAML 字符串字段。5.3 服务启动参数parser.add_argument(--host, default0.0.0.0, helpHost to bind to (default: 0.0.0.0)) parser.add_argument(--port, typeint, default8080, helpPort to bind to (default: 8080))随后以workers1启动 uvicorn。如需换端口或绑定地址python policy_server.py --host 0.0.0.0 --port 9000六、进阶版服务剖析dynamic_policy_server.pydynamic_policy_server.py 展示了更具通用性的两种能力运行时动态选择策略与在 API 处理器内直接实现策略。6.1 通过 --policy 动态加载策略--policy参数接受example_policy包内任意策略类的类名并通过getattr动态获取、校验其确实是sky.AdminPolicy子类后存入app.state.policy_implparser.add_argument(--policy, defaultDoNothingPolicy, helpPolicy to use (default: DoNothingPolicy)) ... policy_class getattr(example_policy, args.policy) assert issubclass(policy_class, sky.AdminPolicy), \ fPolicy {args.policy} is not a valid admin policy app.state.policy_impl policy_class()因此你可以这样切换策略而不改代码# 拒绝所有请求 python dynamic_policy_server.py --policy RejectAllPolicy # 强制 GPU 任务使用 spot 实例 python dynamic_policy_server.py --policy UseSpotForGpuPolicy # 拒绝低版本客户端 python dynamic_policy_server.py --policy RejectOldClientsPolicy对应的处理器与基础版类似但通过request.app.state.policy_impl.apply(user_request)调用逻辑集中在按配置选策略这一处app.post(/) async def apply_policy(request: Request) - JSONResponse: json_data await request.json() user_request sky.UserRequest.decode(json_data) try: mutated_request request.app.state.policy_impl.apply(user_request) except Exception as e: return JSONResponse(contentstr(e), status_code400) return JSONResponse(contentmutated_request.encode())6.2 在 API 处理器内实现策略SetAutostoPolicy演示了就地实现策略的方式——直接操作task.resources的 autostop 配置并独立挂载到/set_autostop端点class SetAutostoPolicy(sky.AdminPolicy): Example: implement a policy at server. classmethod def validate_and_mutate( cls, user_request: sky.UserRequest) - sky.MutatedUserRequest: task user_request.task for r in task.resources: r.override_autostop_config(idle_minutes10) return sky.MutatedUserRequest( tasktask, skypilot_configuser_request.skypilot_config) app.post(/set_autostop) async def set_autostop(request: Request) - JSONResponse: Example: apply the above policy at the API handler. json_data await request.json() user_request sky.UserRequest.decode(json_data) mutated_request SetAutostoPolicy.validate_and_mutate(user_request) return JSONResponse(contentmutated_request.encode())该模式的价值在于你可以为不同请求路径挂不同的策略组合将服务做成策略网关。此外该文件还提供GET /健康检查返回OK便于接入负载均衡或探活。七、API 约定请求与响应格式RESTful 策略服务并不强制使用 FastAPI 或 Python——官方文档指出只要遵循 API 约定任何语言均可实现详见 policy.rst 的 API 约定章节。POST /请求 body 是sky.UserRequest的 JSON 序列化结果典型结构如下摘自 policy.rst{ task: { name: sky-cmd, resources: { cpus: 1 }, num_nodes: 1 }, skypilot_config: {}, request_options: { cluster_name: test, idle_minutes_to_autostop: null, down: false, dryrun: false }, at_client_side: false }其中各字段与 UserRequest 定义 一一对应task用户提交的任务序列化为 YAML 字符串后经 JSON 传输skypilot_config本次请求使用的全局 SkyPilot 配置request_name请求类型如CLUSTER_LAUNCH、CLUSTER_EXEC、OPTIMIZE等定义于sky.server.requests.request_namesrequest_options携带cluster_name、idle_minutes_to_autostop、down、dryrun等附加选项见 RequestOptionsat_client_side请求是否在客户端侧被策略拦截user发起请求的用户信息仅服务端侧可用client_api_version/client_version客户端版本信息仅服务端侧可用用于RejectOldClientsPolicy这类版本管控策略。响应 body 是sky.MutatedUserRequest的 JSON 序列化结果{ task: { name: sky-cmd, resources: { cpus: 1 }, num_nodes: 1 }, skypilot_config: {} }校验失败时返回 HTTP 400body 为错误描述文本。策略可以同时做两件事改写task例如强制use_spot、注入 volume、设置 autostop和改写skypilot_config例如强制aws.use_internal_ip、更新kubernetes.allowed_contexts这赋予了管理员极大的灵活性。八、从示例到生产策略组合与预置策略速览example_policy包预置了丰富的可组合策略全部继承sky.AdminPolicy并实现validate_and_mutate源码见 skypilot_policy.py 与 client_policy.py。以下策略都可在策略服务的policies列表中直接组合复用策略类行为应用侧建议DoNothingPolicy不校验不改写原样放行服务端/客户端RejectAllPolicy拒绝所有请求服务端UseSpotForGpuPolicy所有含 GPU 的任务强制use_spotTrue服务端AddLabelsPolicy为 Kubernetes 任务注入custom_metadata.labels服务端AddLabelsConditionalPolicy仅对 cluster launch 请求注入标签服务端DisablePublicIpPolicyAWS 任务强制内网 IP未配 VPC 则拒绝服务端EnforceAutostopPolicy所有集群必须设置 autostop否则拒绝服务端SetMaxAutostopIdleMinutesPolicy将 autostop 空闲分钟数收敛到上限10 分钟服务端RateLimitLaunchPolicy基于令牌桶对CLUSTER_LAUNCH做用户级限流服务端GPUStaticQuotaPolicy按用户对 H100/L40S 等机型做静态配额校验服务端RejectOldClientsPolicy拒绝 API 版本低于阈值的客户端服务端DynamicKubernetesContextsUpdatePolicy动态刷新 kubeconfig 并收紧allowed_contexts服务端SlurmPartitionRoutingPolicy按资源类型路由到 gpu/highmem/cpu 分区服务端SlurmFilesystemRoutingPolicy按SKYPILOT_REQUIRED_FILESYSTEMS路由到挂载了对应路径的 Slurm 集群服务端OPTIMIZEUseLocalGcpCredentialsPolicy在任务中注入用户本地 GCP 凭证客户端两点注意客户端策略与 RESTful 服务的边界UseLocalGcpCredentialsPolicy是典型的客户端策略——它需要读取用户本机文件~/.config/gcloud而 RESTful 策略运行在服务端无法访问用户本地环境因此该策略不适合放进 RESTful 服务。策略服务里应只放不依赖客户端环境的策略。请求类型感知多个预置策略会通过user_request.request_name与user_request.at_client_side判断生效范围如RateLimitLaunchPolicy只拦CLUSTER_LAUNCH且要求服务端侧组合时无需担心误伤其他请求类型。九、验证与排错9.1 验证策略服务正常# 1) 启动服务 python policy_server.py # 2) 模拟 SkyPilot 客户端发请求POST 一个最小请求体 curl -X POST http://localhost:8080/ \ -H Content-Type: application/json \ -d {task: name: sky-cmd\nresources:\n cpus: \1\\n, skypilot_config: , request_name: CLUSTER_LAUNCH, at_client_side: false, user: }若返回包含task与skypilot_config的 JSON说明编解码链路正常。对dynamic_policy_server.py还可访问GET /做健康检查返回OK。9.2 端到端验证策略生效在 SkyPilot 配置中写入admin_policy: http://localhost:8080参考 restful_policy.yaml后执行sky launch服务端侧应用策略时会打印Applying server admin policy: RestfulAdminPolicy(policy_urlhttp://localhost:8080)日志位置见 admin_policy_utils.py 的 apply 函数。若策略拒绝请求客户端会收到红色提示User request is rejected by admin policy http://localhost:8080: ...参见 RestfulAdminPolicy 的异常处理。9.3 常见问题Failed to call admin policy URLSkyPilot 客户端或 API Server无法连接策略服务检查服务是否启动、--host是否可被客户端访问、防火墙与 30 秒超时是否足够Failed to decode response服务返回的 body 不符合MutatedUserRequestJSON 结构通常是自定义实现未正确输出task/skypilot_config两个字段策略未生效确认admin_policy键写在正确的配置层级服务端侧策略写在 API Server 的配置中而非仅写在用户本地并确认policy没有因at_client_side/request_name条件被跳过--policy参数报错policy_server.py当前实现不接受--policy若需命令行动态选策略请使用dynamic_policy_server.py见第三节的对比表。十、进一步阅读官方 Admin Policy 完整文档部署、实现、示例策略详解docs/source/cloud-setup/policy.rst策略接口与数据类型定义AdminPolicy、UserRequest、MutatedUserRequest、RestfulAdminPolicysky/admin_policy.py策略加载与调用编排URL 识别、策略应用、配置覆盖sky/utils/admin_policy_utils.py预置策略源码与导出清单example_policy/example_policy/skypilot_policy.py、example_policy/example_policy/init.py各策略配套的配置示例reject_all.yaml、use_spot_for_gpu.yaml、gpu_static_quota.yaml、reject_old_clients.yaml等examples/admin_policy通过本文的示例服务你可以快速在组织内部落地统一策略裁决的治理模型把配额、限流、成本控制、合规校验集中到一个 RESTful 服务中SkyPilot 的所有用户请求在进入调度前都经过统一的策略管道从而实现对整个多租户平台的可审计、可管控治理。【免费下载链接】skypilotThe AI Compute Platform for frontier teams. SkyPilot turns fragmented AI compute into one AI supercomputer, so frontier AI teams build custom intelligence faster.项目地址: https://gitcode.com/GitHub_Trending/sk/skypilot创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 SkyPilot 管理策略服务实战：用 FastAPI 构建 RESTful Admin Policy Server
SkyPilot 管理策略服务实战用 FastAPI 构建 RESTful Admin Policy Server【免费下载链接】skypilotThe AI Compute Platform for frontier teams. SkyPilot turns fragmented AI compute into one AI supercomputer, so frontier AI teams build custom intelligence faster.项目地址: https://gitcode.com/GitHub_Trending/sk/skypilotSkyPilot 的 Admin Policy管理策略机制允许平台管理员对所有用户的sky launch/sky exec等请求执行自定义的校验与改写逻辑。本指南以仓库 examples/admin_policy/example_server 为例完整讲解如何用 FastAPI 将一个或多个 Admin Policy 封装成独立的 RESTful 策略服务并通过admin_policy: http://localhost:8080接入 SkyPilot。读完本文你将掌握 RESTful 策略服务的完整生命周期理解UserRequest/MutatedUserRequest的编解码约定、读懂示例服务的核心实现、学会动态加载外部策略包并能为自己的组织定制可独立部署的策略服务。一、为什么需要 RESTful 形式的 Admin PolicySkyPilot 采用客户端—服务端架构一个集中式的 API Server 可以统一承载用户的请求管理员则通过 Admin Policy 在请求进入核心流程之前注入校验 改写逻辑。在 docs/source/cloud-setup/policy.rst 中官方将部署方式划分为两类Python 包方式把实现sky.AdminPolicy的类发布为 Python 包安装到 SkyPilot 所在的 Python 环境并在配置中指定admin_policy: mypackage.MyPolicyRESTful 服务方式把策略逻辑托管在一个独立 HTTP 服务中配置中直接填写 URL例如admin_policy: https://example.com/policy。RESTful 方式的核心优势在于策略逻辑与 SkyPilot 环境完全解耦策略可以用任意语言实现只要遵循 API 约定可以集中部署、热更新且天然适合多用户、多客户端的场景——所有用户请求都汇聚到同一个策略端点统一裁决。代价是策略运行在服务端所在的主机无法访问发起请求的用户的本地环境如本地文件、本地凭证这一点官方文档也明确提示参见 policy.rst 中的说明。本文涉及的example_server目录正是这种 RESTful 方式的官方参考实现examples/admin_policy/example_server/ ├── README.md # 本指南对应的原始文档 ├── policy_server.py # 基础版内置策略组合 FastAPI 服务 └── dynamic_policy_server.py# 进阶版支持 --policy 动态加载外部策略二、准备工作与依赖2.1 安装 SkyPilot策略服务的核心数据类型sky.UserRequest、sky.MutatedUserRequest、sky.AdminPolicy都来自 SkyPilot 本身因此必须先安装 SkyPilotpip install skypilot安装完成后import sky即可获得策略接口与请求对象接口定义见 sky/admin_policy.py。2.2 安装示例策略包示例服务会引用example_policy包中预置的策略类如UseSpotForGpuPolicy。该包位于仓库 examples/admin_policy/example_policy安装方式如下在克隆仓库后执行cd skypilot pip install examples/admin_policy/example_policy该包通过 pyproject.toml 声明name example_policy、version 0.0.1并在init.py 中导出全部预置策略包括DoNothingPolicy、UseSpotForGpuPolicy、RejectAllPolicy、AddLabelsPolicy、DisablePublicIpPolicy、EnforceAutostopPolicy、RateLimitLaunchPolicy、GPUStaticQuotaPolicy、RejectOldClientsPolicy等。2.3 服务端运行时依赖示例服务基于 FastAPI 与 uvicorn 编写启动前需确保已安装pip install fastapi uvicorn三、启动策略服务按 README 的 Quickstart启动命令为python policy_server.py --policy DoNothingPolicy服务默认监听0.0.0.0:8080。这里需要说明一个仓库内的实现细节当前仓库中真正支持--policy参数的是 dynamic_policy_server.py而 policy_server.py 的 argparse 只暴露了--host与--port策略组合直接硬编码在policies列表中。两个文件均可独立启动服务区别见下文第五、六节脚本策略来源额外端点policy_server.py硬编码组合DoNothingPolicyUseSpotForGpuPolicyPOST /dynamic_policy_server.py--policy动态加载example_policy中的任意策略GET /健康检查、POST /set_autostop四、把策略服务接入 SkyPilot服务启动后在 SkyPilot 配置中设置admin_policy字段指向该服务admin_policy: http://localhost:8080仓库 examples/admin_policy/restful_policy.yaml 提供的正是这一最小配置。之后执行sky launch时SkyPilot 会先调用该 URL 对请求做校验与改写再继续后续流程。4.1 SkyPilot 如何识别并调用 URL 策略从源码看URL 与 Python 包路径的区分发生在 sky/utils/admin_policy_utils.py 的_get_policy_impl中_is_url()通过urlparse判断policy_string的 scheme 是否为http/https若为 URL则构造内置的RestfulAdminPolicy(policy_location)否则按module_path.class_name的格式importlib动态导入策略类。4.2 RestfulAdminPolicy 的调用约定RESTful 调用端的实现RestfulAdminPolicy完整定义了客户端与服务端的契约向policy_url发起POST请求body 为user_request.encode()的 JSON 字符串Content-Type: application/json默认超时 30 秒收到 HTTP 400 时抛出UserRequestRejectedByPolicy即策略主动拒绝请求其他非 2xx 状态码触发response.raise_for_status()网络类异常统一包装为RestfulPolicyError成功时用MutatedUserRequest.decode(response.json(), user_request)反序列化改写后的请求并恢复内部字段managed_job_dag、service_name。这意味策略服务的响应必须符合MutatedUserRequest的 JSON 格式否则会报Failed to decode response from admin policy URL。五、基础版服务剖析policy_server.pypolicy_server.py 是 README 指向的核心实现完整展示了请求进来 → 解码 → 逐策略应用 → 编码返回的完整链路。5.1 在服务内实现一个策略DoNothingPolicy是最简单的sky.AdminPolicy子类——不校验、不改写原样返回任务与配置class DoNothingPolicy(sky.AdminPolicy): Example policy: do nothing. classmethod def validate_and_mutate( cls, user_request: sky.UserRequest) - sky.MutatedUserRequest: Returns the user request unchanged. return sky.MutatedUserRequest(user_request.task, user_request.skypilot_config)这正是 AdminPolicy 抽象接口 要求的唯一抽象方法签名validate_and_mutate(cls, user_request) - MutatedUserRequest。需要拒绝请求时策略内raise异常即可服务端会统一转为 HTTP 400。5.2 唯一的策略端点POST /apply_policy处理器是服务的心脏app.post(/) async def apply_policy(request: Request) - JSONResponse: Apply an admin policy loaded from external package to a user request # Decode json_data await request.json() user_request sky.UserRequest.decode(json_data) # Example: change the following list to apply different policies. policies: List[sky.AdminPolicy] [ # Example: policy that implemented in the server package. DoNothingPolicy, # Example: policy from third party packages. example_policy.UseSpotForGpuPolicy, ] try: for policy in policies: mutated_request policy.validate_and_mutate(user_request) user_request.task mutated_request.task user_request.skypilot_config mutated_request.skypilot_config except Exception as e: # pylint: disablebroad-except return JSONResponse(contentstr(e), status_code400) return JSONResponse(contentmutated_request.encode())要点拆解解码sky.UserRequest.decode(json_data)将请求 body 还原为UserRequest内部通过 pydantic 模型_UserRequestBody校验再把 YAML 字符串反序列化为sky.Task与sky.Config见 sky/admin_policy.py组合策略policies列表是策略管道pipeline按顺序逐个执行前一个策略改写后的task/skypilot_config会覆盖回user_request供下一个策略继续处理。这里展示了两种来源的混用服务内实现的DoNothingPolicy 外部包example_policy提供的UseSpotForGpuPolicy拒绝语义任一策略抛异常整个请求被拒绝响应 400 并携带异常文本——与 SkyPilot 客户端 400 →UserRequestRejectedByPolicy的处理完全对应编码mutated_request.encode()输出符合_MutatedUserRequestBody结构的 JSONtaskskypilot_config两个 YAML 字符串字段。5.3 服务启动参数parser.add_argument(--host, default0.0.0.0, helpHost to bind to (default: 0.0.0.0)) parser.add_argument(--port, typeint, default8080, helpPort to bind to (default: 8080))随后以workers1启动 uvicorn。如需换端口或绑定地址python policy_server.py --host 0.0.0.0 --port 9000六、进阶版服务剖析dynamic_policy_server.pydynamic_policy_server.py 展示了更具通用性的两种能力运行时动态选择策略与在 API 处理器内直接实现策略。6.1 通过 --policy 动态加载策略--policy参数接受example_policy包内任意策略类的类名并通过getattr动态获取、校验其确实是sky.AdminPolicy子类后存入app.state.policy_implparser.add_argument(--policy, defaultDoNothingPolicy, helpPolicy to use (default: DoNothingPolicy)) ... policy_class getattr(example_policy, args.policy) assert issubclass(policy_class, sky.AdminPolicy), \ fPolicy {args.policy} is not a valid admin policy app.state.policy_impl policy_class()因此你可以这样切换策略而不改代码# 拒绝所有请求 python dynamic_policy_server.py --policy RejectAllPolicy # 强制 GPU 任务使用 spot 实例 python dynamic_policy_server.py --policy UseSpotForGpuPolicy # 拒绝低版本客户端 python dynamic_policy_server.py --policy RejectOldClientsPolicy对应的处理器与基础版类似但通过request.app.state.policy_impl.apply(user_request)调用逻辑集中在按配置选策略这一处app.post(/) async def apply_policy(request: Request) - JSONResponse: json_data await request.json() user_request sky.UserRequest.decode(json_data) try: mutated_request request.app.state.policy_impl.apply(user_request) except Exception as e: return JSONResponse(contentstr(e), status_code400) return JSONResponse(contentmutated_request.encode())6.2 在 API 处理器内实现策略SetAutostoPolicy演示了就地实现策略的方式——直接操作task.resources的 autostop 配置并独立挂载到/set_autostop端点class SetAutostoPolicy(sky.AdminPolicy): Example: implement a policy at server. classmethod def validate_and_mutate( cls, user_request: sky.UserRequest) - sky.MutatedUserRequest: task user_request.task for r in task.resources: r.override_autostop_config(idle_minutes10) return sky.MutatedUserRequest( tasktask, skypilot_configuser_request.skypilot_config) app.post(/set_autostop) async def set_autostop(request: Request) - JSONResponse: Example: apply the above policy at the API handler. json_data await request.json() user_request sky.UserRequest.decode(json_data) mutated_request SetAutostoPolicy.validate_and_mutate(user_request) return JSONResponse(contentmutated_request.encode())该模式的价值在于你可以为不同请求路径挂不同的策略组合将服务做成策略网关。此外该文件还提供GET /健康检查返回OK便于接入负载均衡或探活。七、API 约定请求与响应格式RESTful 策略服务并不强制使用 FastAPI 或 Python——官方文档指出只要遵循 API 约定任何语言均可实现详见 policy.rst 的 API 约定章节。POST /请求 body 是sky.UserRequest的 JSON 序列化结果典型结构如下摘自 policy.rst{ task: { name: sky-cmd, resources: { cpus: 1 }, num_nodes: 1 }, skypilot_config: {}, request_options: { cluster_name: test, idle_minutes_to_autostop: null, down: false, dryrun: false }, at_client_side: false }其中各字段与 UserRequest 定义 一一对应task用户提交的任务序列化为 YAML 字符串后经 JSON 传输skypilot_config本次请求使用的全局 SkyPilot 配置request_name请求类型如CLUSTER_LAUNCH、CLUSTER_EXEC、OPTIMIZE等定义于sky.server.requests.request_namesrequest_options携带cluster_name、idle_minutes_to_autostop、down、dryrun等附加选项见 RequestOptionsat_client_side请求是否在客户端侧被策略拦截user发起请求的用户信息仅服务端侧可用client_api_version/client_version客户端版本信息仅服务端侧可用用于RejectOldClientsPolicy这类版本管控策略。响应 body 是sky.MutatedUserRequest的 JSON 序列化结果{ task: { name: sky-cmd, resources: { cpus: 1 }, num_nodes: 1 }, skypilot_config: {} }校验失败时返回 HTTP 400body 为错误描述文本。策略可以同时做两件事改写task例如强制use_spot、注入 volume、设置 autostop和改写skypilot_config例如强制aws.use_internal_ip、更新kubernetes.allowed_contexts这赋予了管理员极大的灵活性。八、从示例到生产策略组合与预置策略速览example_policy包预置了丰富的可组合策略全部继承sky.AdminPolicy并实现validate_and_mutate源码见 skypilot_policy.py 与 client_policy.py。以下策略都可在策略服务的policies列表中直接组合复用策略类行为应用侧建议DoNothingPolicy不校验不改写原样放行服务端/客户端RejectAllPolicy拒绝所有请求服务端UseSpotForGpuPolicy所有含 GPU 的任务强制use_spotTrue服务端AddLabelsPolicy为 Kubernetes 任务注入custom_metadata.labels服务端AddLabelsConditionalPolicy仅对 cluster launch 请求注入标签服务端DisablePublicIpPolicyAWS 任务强制内网 IP未配 VPC 则拒绝服务端EnforceAutostopPolicy所有集群必须设置 autostop否则拒绝服务端SetMaxAutostopIdleMinutesPolicy将 autostop 空闲分钟数收敛到上限10 分钟服务端RateLimitLaunchPolicy基于令牌桶对CLUSTER_LAUNCH做用户级限流服务端GPUStaticQuotaPolicy按用户对 H100/L40S 等机型做静态配额校验服务端RejectOldClientsPolicy拒绝 API 版本低于阈值的客户端服务端DynamicKubernetesContextsUpdatePolicy动态刷新 kubeconfig 并收紧allowed_contexts服务端SlurmPartitionRoutingPolicy按资源类型路由到 gpu/highmem/cpu 分区服务端SlurmFilesystemRoutingPolicy按SKYPILOT_REQUIRED_FILESYSTEMS路由到挂载了对应路径的 Slurm 集群服务端OPTIMIZEUseLocalGcpCredentialsPolicy在任务中注入用户本地 GCP 凭证客户端两点注意客户端策略与 RESTful 服务的边界UseLocalGcpCredentialsPolicy是典型的客户端策略——它需要读取用户本机文件~/.config/gcloud而 RESTful 策略运行在服务端无法访问用户本地环境因此该策略不适合放进 RESTful 服务。策略服务里应只放不依赖客户端环境的策略。请求类型感知多个预置策略会通过user_request.request_name与user_request.at_client_side判断生效范围如RateLimitLaunchPolicy只拦CLUSTER_LAUNCH且要求服务端侧组合时无需担心误伤其他请求类型。九、验证与排错9.1 验证策略服务正常# 1) 启动服务 python policy_server.py # 2) 模拟 SkyPilot 客户端发请求POST 一个最小请求体 curl -X POST http://localhost:8080/ \ -H Content-Type: application/json \ -d {task: name: sky-cmd\nresources:\n cpus: \1\\n, skypilot_config: , request_name: CLUSTER_LAUNCH, at_client_side: false, user: }若返回包含task与skypilot_config的 JSON说明编解码链路正常。对dynamic_policy_server.py还可访问GET /做健康检查返回OK。9.2 端到端验证策略生效在 SkyPilot 配置中写入admin_policy: http://localhost:8080参考 restful_policy.yaml后执行sky launch服务端侧应用策略时会打印Applying server admin policy: RestfulAdminPolicy(policy_urlhttp://localhost:8080)日志位置见 admin_policy_utils.py 的 apply 函数。若策略拒绝请求客户端会收到红色提示User request is rejected by admin policy http://localhost:8080: ...参见 RestfulAdminPolicy 的异常处理。9.3 常见问题Failed to call admin policy URLSkyPilot 客户端或 API Server无法连接策略服务检查服务是否启动、--host是否可被客户端访问、防火墙与 30 秒超时是否足够Failed to decode response服务返回的 body 不符合MutatedUserRequestJSON 结构通常是自定义实现未正确输出task/skypilot_config两个字段策略未生效确认admin_policy键写在正确的配置层级服务端侧策略写在 API Server 的配置中而非仅写在用户本地并确认policy没有因at_client_side/request_name条件被跳过--policy参数报错policy_server.py当前实现不接受--policy若需命令行动态选策略请使用dynamic_policy_server.py见第三节的对比表。十、进一步阅读官方 Admin Policy 完整文档部署、实现、示例策略详解docs/source/cloud-setup/policy.rst策略接口与数据类型定义AdminPolicy、UserRequest、MutatedUserRequest、RestfulAdminPolicysky/admin_policy.py策略加载与调用编排URL 识别、策略应用、配置覆盖sky/utils/admin_policy_utils.py预置策略源码与导出清单example_policy/example_policy/skypilot_policy.py、example_policy/example_policy/init.py各策略配套的配置示例reject_all.yaml、use_spot_for_gpu.yaml、gpu_static_quota.yaml、reject_old_clients.yaml等examples/admin_policy通过本文的示例服务你可以快速在组织内部落地统一策略裁决的治理模型把配额、限流、成本控制、合规校验集中到一个 RESTful 服务中SkyPilot 的所有用户请求在进入调度前都经过统一的策略管道从而实现对整个多租户平台的可审计、可管控治理。【免费下载链接】skypilotThe AI Compute Platform for frontier teams. SkyPilot turns fragmented AI compute into one AI supercomputer, so frontier AI teams build custom intelligence faster.项目地址: https://gitcode.com/GitHub_Trending/sk/skypilot创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考