Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 panxiao81
|
09a0a5665f
|
||
|
panxiao81
|
d9d99b2f9b
|
||
|
panxiao81
|
3e21de5942
|
||
|
panxiao81
|
f76978ca17
|
||
|
panxiao81
|
06bc54e3cf
|
||
|
panxiao81
|
86effb72a8
|
@@ -1,76 +0,0 @@
|
|||||||
---
|
|
||||||
name: homelab-knowledge
|
|
||||||
description: Query and maintain the shared homelab-wiki when working on homelab services, infrastructure, architecture, operations, or current service status. Use it to gather existing context before work and to keep durable knowledge synchronized after relevant changes; do not use it for unrelated software work or as a substitute for commit and PR history.
|
|
||||||
---
|
|
||||||
|
|
||||||
# Homelab Knowledge
|
|
||||||
|
|
||||||
Use `homelab-wiki` as the shared long-lived knowledge base for people and agents. Search it directly with `rg`; do not introduce a search index, vector database, or generated copy of the wiki.
|
|
||||||
|
|
||||||
## Locate the wiki
|
|
||||||
|
|
||||||
Resolve the checkout in this order:
|
|
||||||
|
|
||||||
1. `$HOMELAB_WIKI_PATH`, when set.
|
|
||||||
2. A sibling directory named `homelab-wiki` next to the current repository.
|
|
||||||
3. `/home/panxiao81/homelab-wiki` when it exists.
|
|
||||||
|
|
||||||
If no checkout is available, report that constraint. Do not silently skip the knowledge step, clone a repository, or create a replacement wiki without the user's authorization.
|
|
||||||
|
|
||||||
Before using the wiki, read its `AGENTS.md` completely. For edits, also read `README.md` and `CONTRIBUTING.md` completely and follow any more specific instructions associated with the target page.
|
|
||||||
|
|
||||||
## Gather context
|
|
||||||
|
|
||||||
At the beginning of a homelab task:
|
|
||||||
|
|
||||||
1. Derive search terms from the component name, service aliases, hostnames, Kubernetes resources, configuration keys, error text, and task intent.
|
|
||||||
2. Use `rg -n -i` in the wiki to find candidate pages. Prefer several precise searches over reading the whole repository.
|
|
||||||
3. Follow the wiki's task index, service index, architecture constraints, source records, and verification conflicts when they are relevant.
|
|
||||||
4. Read the closest authoritative pages and their material links before making decisions. Also read the corresponding source repository README or runbook when changing an implementation.
|
|
||||||
5. Distinguish documented design, declared configuration, deployment history, live verification, and work currently in progress. Do not present one as another.
|
|
||||||
|
|
||||||
For questions about current project or service status, first obtain the maintainer's current-work and ticket context as required by the wiki, unless the conversation already provides that authorization and scope. Reading documentation does not authorize live-system inspection.
|
|
||||||
|
|
||||||
Answer read-only questions from the evidence found. Include paths or links that let the user verify important claims, and state when evidence may be stale or conflicting.
|
|
||||||
|
|
||||||
## Maintain knowledge after changes
|
|
||||||
|
|
||||||
For any code, configuration, infrastructure, or operational change, perform a documentation-impact check before declaring the task complete.
|
|
||||||
|
|
||||||
Update the wiki in the same task when the change affects durable knowledge such as:
|
|
||||||
|
|
||||||
- service purpose, lifecycle, entry point, authentication, permissions, dependencies, or first-use path;
|
|
||||||
- architecture boundaries or accepted constraints;
|
|
||||||
- deployment ownership or persistent operating behavior;
|
|
||||||
- troubleshooting, recovery, verification, or maintenance procedures;
|
|
||||||
- the addition, replacement, or retirement of a service.
|
|
||||||
|
|
||||||
Keep one-time progress, implementation narration, and release-by-release history in commits, PRs, or tickets. Do not copy them into the wiki unless they change a durable stage summary. Implementation-specific parameters may remain in the source repository README or runbook when the wiki convention says to link rather than duplicate them.
|
|
||||||
|
|
||||||
When editing:
|
|
||||||
|
|
||||||
1. Inspect both the source-repository diff and the wiki working tree before writing. Preserve unrelated user changes in both repositories.
|
|
||||||
2. Update the page closest to the fact first, then only the navigation, indexes, constraints, or verification records that the wiki rules require.
|
|
||||||
3. Preserve evidence metadata. Never advance `last_verified` without performing the stated live verification; ordinary review may update only fields permitted by the wiki.
|
|
||||||
4. Link related source commits, PRs, or paths when available. Clearly mark uncommitted sources and unfinished cross-repository synchronization.
|
|
||||||
5. Record conflicts rather than resolving them by assumption. Ask before live inspection or before choosing among materially conflicting current-state claims.
|
|
||||||
6. Keep credentials, tokens, private keys, Terraform state, secret values, and sensitive command output out of documentation. Never read or copy known sensitive files merely to improve the wiki.
|
|
||||||
|
|
||||||
Wiki edits are a separate repository change. Do not commit, push, open a PR, or modify a live system unless the user has authorized that action.
|
|
||||||
|
|
||||||
## Verify and report
|
|
||||||
|
|
||||||
After editing the wiki, run from its root:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
python3 scripts/check_docs.py
|
|
||||||
git diff --check
|
|
||||||
```
|
|
||||||
|
|
||||||
If the checker itself changed, also run:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
python3 -m unittest discover -s tests -v
|
|
||||||
```
|
|
||||||
|
|
||||||
In the final response, report source-repository changes and wiki changes separately, including validation performed and anything still awaiting verification or cross-repository linkage. If no wiki update was needed, state the concrete reason; do not merely say that documentation was unaffected.
|
|
||||||
@@ -32,8 +32,6 @@ Ayatori 是 `ddupan.top` homelab 的内部基础设施控制平面。它以 Kube
|
|||||||
- [ADR-0002：采用 k0s 与可选工作负载运行时](docs/decisions/0002-k0s-optional-workload-runtime.md)
|
- [ADR-0002：采用 k0s 与可选工作负载运行时](docs/decisions/0002-k0s-optional-workload-runtime.md)
|
||||||
- [ADR-0003：直接连接 Dev API 的开发循环](docs/decisions/0003-dev-api-development-loop.md)
|
- [ADR-0003：直接连接 Dev API 的开发循环](docs/decisions/0003-dev-api-development-loop.md)
|
||||||
- [ADR-0006：按实际管理缺口扩展资源 API](docs/decisions/0006-demand-driven-resource-scope.md)
|
- [ADR-0006：按实际管理缺口扩展资源 API](docs/decisions/0006-demand-driven-resource-scope.md)
|
||||||
- [ADR-0007：复用 Node API 建立按需实现的 Compute 能力](docs/decisions/0007-compute-node-and-vm-boundary.md)
|
|
||||||
- [ADR-0008：将 PostgreSQL Tenant Operator 合并为 Ayatori Database 模块](docs/decisions/0008-merge-postgresql-tenant-operator.md)
|
|
||||||
|
|
||||||
## 当前状态
|
## 当前状态
|
||||||
|
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ import (
|
|||||||
"sigs.k8s.io/controller-runtime/pkg/webhook"
|
"sigs.k8s.io/controller-runtime/pkg/webhook"
|
||||||
|
|
||||||
executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
|
executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
|
||||||
|
"git.ddupan.top/panxiao81/ayatori/internal/controller"
|
||||||
// +kubebuilder:scaffold:imports
|
// +kubebuilder:scaffold:imports
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -165,6 +166,13 @@ func main() {
|
|||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err := (&controller.JobReconciler{
|
||||||
|
Client: mgr.GetClient(),
|
||||||
|
}).SetupWithManager(mgr); err != nil {
|
||||||
|
setupLog.Error(err, "Failed to create controller", "controller", "Job")
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
|
|
||||||
// +kubebuilder:scaffold:builder
|
// +kubebuilder:scaffold:builder
|
||||||
|
|
||||||
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
|
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
|
||||||
|
|||||||
+53
-6
@@ -1,11 +1,58 @@
|
|||||||
|
---
|
||||||
apiVersion: rbac.authorization.k8s.io/v1
|
apiVersion: rbac.authorization.k8s.io/v1
|
||||||
kind: ClusterRole
|
kind: ClusterRole
|
||||||
metadata:
|
metadata:
|
||||||
labels:
|
|
||||||
app.kubernetes.io/name: ayatori
|
|
||||||
app.kubernetes.io/managed-by: kustomize
|
|
||||||
name: manager-role
|
name: manager-role
|
||||||
rules:
|
rules:
|
||||||
- apiGroups: [""]
|
- apiGroups:
|
||||||
resources: ["pods"]
|
- ""
|
||||||
verbs: ["get", "list", "watch"]
|
resources:
|
||||||
|
- namespaces
|
||||||
|
- serviceaccounts
|
||||||
|
verbs:
|
||||||
|
- get
|
||||||
|
- list
|
||||||
|
- watch
|
||||||
|
- apiGroups:
|
||||||
|
- batch
|
||||||
|
resources:
|
||||||
|
- jobs
|
||||||
|
verbs:
|
||||||
|
- create
|
||||||
|
- delete
|
||||||
|
- get
|
||||||
|
- list
|
||||||
|
- watch
|
||||||
|
- apiGroups:
|
||||||
|
- execution.ayatori.ddupan.top
|
||||||
|
resources:
|
||||||
|
- jobclasses
|
||||||
|
- kubernetesexecutionparameters
|
||||||
|
verbs:
|
||||||
|
- get
|
||||||
|
- list
|
||||||
|
- watch
|
||||||
|
- apiGroups:
|
||||||
|
- execution.ayatori.ddupan.top
|
||||||
|
resources:
|
||||||
|
- jobs
|
||||||
|
verbs:
|
||||||
|
- get
|
||||||
|
- list
|
||||||
|
- patch
|
||||||
|
- update
|
||||||
|
- watch
|
||||||
|
- apiGroups:
|
||||||
|
- execution.ayatori.ddupan.top
|
||||||
|
resources:
|
||||||
|
- jobs/finalizers
|
||||||
|
verbs:
|
||||||
|
- update
|
||||||
|
- apiGroups:
|
||||||
|
- execution.ayatori.ddupan.top
|
||||||
|
resources:
|
||||||
|
- jobs/status
|
||||||
|
verbs:
|
||||||
|
- get
|
||||||
|
- patch
|
||||||
|
- update
|
||||||
|
|||||||
@@ -1,61 +0,0 @@
|
|||||||
# ADR-0007：复用 Node API 建立按需实现的 Compute 能力
|
|
||||||
|
|
||||||
- 状态：Accepted
|
|
||||||
- 日期：2026-09-20
|
|
||||||
- 实施优先级：Deferred；当前优先 Database、LoadBalancer 与 Bucket
|
|
||||||
|
|
||||||
## 背景
|
|
||||||
|
|
||||||
Ayatori 长期可能需要管理现有 Proxmox VM、当前 libvirt VM，以及允许普通计算节点临时加入、
|
|
||||||
排空和退出。Proxmox 的远程 API 不能覆盖全部所需操作；若 Ayatori 进一步实现节点 inventory、
|
|
||||||
简单 placement、fencing 和安全 reschedule，Proxmox 的控制面价值会逐步被替代。
|
|
||||||
|
|
||||||
同一物理节点未来也可能运行 OpenSandbox/Kata 等执行后端。Kata 虽然以 microVM 隔离 Pod 或
|
|
||||||
container，但其公开生命周期是 Sandbox/Run，不是具有磁盘、NIC、console、placement、迁移和
|
|
||||||
长期身份的 VirtualMachine 产品。
|
|
||||||
|
|
||||||
## 决策
|
|
||||||
|
|
||||||
### 节点 API
|
|
||||||
|
|
||||||
Ayatori 选择性复用 `core/v1 Node` 与 `coordination.k8s.io/v1 Lease` 表达计算节点身份、能力、
|
|
||||||
容量、健康、维护状态与心跳。它们只是 API contract：由 Ayatori Compute Agent 写入，并由
|
|
||||||
Ayatori 自有 controller 消费。
|
|
||||||
|
|
||||||
这项选择不引入 kubelet、Pod、CRI、kube-scheduler 或 kube-controller-manager。Compute Agent
|
|
||||||
不是对 kubelet 的模拟或兼容实现，而是 Node API 在 Ayatori Compute 领域中的正式 producer。
|
|
||||||
每个 Node 必须带 Ayatori ownership label；Agent 只能更新自己的 Node/status 与 Lease。
|
|
||||||
|
|
||||||
初版 VirtualMachine 显式指定 Node。出现实际需求后，再由 Ayatori controller 基于 Node 的
|
|
||||||
Ready、unschedulable、taints、labels、capacity 和已有 allocation 实现小规模 filter/score。
|
|
||||||
具体资源分配不能依靠多个 controller 反复改写 `Node.status.allocatable`；需要并发预留时增加
|
|
||||||
独立 Allocation 资源或等价的原子分配记录。
|
|
||||||
|
|
||||||
### VM 数据面
|
|
||||||
|
|
||||||
长期主路径可以是普通 Linux Compute Node 上的 libvirt/QEMU，由受限的 Compute Agent 执行
|
|
||||||
版本化、强类型、幂等且可观察的 VM 操作。Agent 不提供任意远程 shell。
|
|
||||||
|
|
||||||
Proxmox 是 brownfield 迁移后端：初期用于 adopt 现有 VM，并继续提供当前已有的集群、存储、
|
|
||||||
备份与 HA 能力。若 Ayatori Compute 已经可靠覆盖所需 placement、fencing、存储可移植性和恢复
|
|
||||||
语义，可以逐步把 PVE 节点迁移为普通 Compute Node；不为维持虚假 backend 对等性承诺永久支持
|
|
||||||
所有 Proxmox 特性。
|
|
||||||
|
|
||||||
### HA 边界
|
|
||||||
|
|
||||||
自动 reschedule 必须满足：旧节点已经可靠 fenced，且 Volume 明确报告可在目标节点使用。
|
|
||||||
任一条件无法证明时，VM 进入 Blocked/ManualTask，不得冒险在第二个节点启动。首版允许完全
|
|
||||||
人工 placement 与恢复；不以通用 Placement、透明 live migration、多租户 SDN 或 Nova 兼容为目标。
|
|
||||||
|
|
||||||
### Sandbox 边界
|
|
||||||
|
|
||||||
OpenSandbox/Kata microVM 归属于 Run/Sandbox backend 的隔离实现，不创建 VirtualMachine 资源。
|
|
||||||
若未来 VM 与 Sandbox 共享物理节点，容量协调必须另行形成经过验证的设计；不能仅因两者底层
|
|
||||||
都使用 KVM 就合并其北向生命周期。
|
|
||||||
|
|
||||||
## 结果
|
|
||||||
|
|
||||||
- 复用成熟 Node/Lease API，而不继承 Kubernetes workload plane。
|
|
||||||
- Compute 能力可以按 homelab 所需规模实现，不必复制完整 Nova。
|
|
||||||
- PVE 帮助现有资源平滑迁移，但不是长期架构必须保留的一层。
|
|
||||||
- Compute 方向已记录，但不改变当前 Database、LoadBalancer、Bucket 的产品优先级。
|
|
||||||
@@ -1,92 +0,0 @@
|
|||||||
# ADR-0008：将 PostgreSQL Tenant Operator 合并为 Ayatori Database 模块
|
|
||||||
|
|
||||||
- 状态：Accepted
|
|
||||||
- 日期：2026-09-20
|
|
||||||
|
|
||||||
## 背景
|
|
||||||
|
|
||||||
独立仓库 `postgresql-tenant-operator` 已经为 homelab 共享 PostgreSQL 设计了
|
|
||||||
`PostgreSQLInstance` 与 `PostgreSQLTenant` API，并包含批准的行为规格、领域值对象、状态机、
|
|
||||||
PostgreSQL ownership registry、OpenBao/External Secrets 边界、迁移与恢复文档及测试。
|
|
||||||
|
|
||||||
Database 是 Ayatori 当前最优先的真实管理缺口之一。继续把该 controller 作为独立产品，会重复
|
|
||||||
维护 manager、API machinery、发布、认证、可观测性和通用 controller 约定，也会使后续应用组合
|
|
||||||
必须跨两个控制平面理解状态。
|
|
||||||
|
|
||||||
截至 2026-09-20，源仓库已经合并 Instance 的 Endpoint、凭据引用、身份/版本、定义与观测目标
|
|
||||||
等值对象，以及扩展支持模型和最小生命周期/checkpoint。它们尚未接入实际运行链路。完整 Ready
|
|
||||||
判定、Kubernetes Secret 管理凭据与连接刷新、应用层/数据库 adapter/controller 接入、CRD 规格
|
|
||||||
对齐及集成验证仍未完成；Tenant 的创建、凭据交付与 Retain/Delete 生命周期也未落地。
|
|
||||||
|
|
||||||
现有运行链路仍是直接读取 OpenBao 管理凭据的旧实现，不能作为新设计已经可用的证据。源仓库
|
|
||||||
本地 `feature/instance-extension-observations` 还保留两个未提交文件，用于 Instance 接受扩展观测
|
|
||||||
及测试；该工作已暂停，不能作为已合并能力或迁移基线。部分生成的 CRD/API 代码也仍落后于批准
|
|
||||||
规范，因此迁移不能把当前工作树或全部脚手架原样复制到 Ayatori。
|
|
||||||
|
|
||||||
## 决策
|
|
||||||
|
|
||||||
PostgreSQL Tenant Operator 合并为 Ayatori 的 Database 领域模块。保留已经批准且仍适用的安全、
|
|
||||||
所有权、幂等与删除行为，不重新发明 database、role、credential 和 registry 语义。
|
|
||||||
|
|
||||||
当前没有可用发布版本、没有被该 operator 托管的 PostgreSQL 实例或 Tenant，也没有需要在线
|
|
||||||
转换的已部署 CR。因此此次合并不承担旧实现兼容性：旧运行链路可以直接撤销，不保留直接读取
|
|
||||||
OpenBao 管理凭据的路径，也不兼容落后于规范的旧 CRD、samples 或实现细节。
|
|
||||||
|
|
||||||
没有部署兼容负担不等于重新设计已经批准的产品合同。源项目的系统规格、API 语义、Instance 与
|
|
||||||
Tenant 领域模型、状态机、ownership registry、OpenBao/ExternalSecret 凭据交付、Retain/Delete、
|
|
||||||
恢复与测试设计整体作为 Ayatori Database 模块的规范基线。除 API group、项目归属和装配结构外，
|
|
||||||
迁移不得静默改变这些行为；确需改变时必须先单独修订规格并记录决定。
|
|
||||||
|
|
||||||
目标结构遵守 Ayatori 的模块化单体边界：
|
|
||||||
|
|
||||||
```text
|
|
||||||
api/database/v1alpha1/
|
|
||||||
internal/database/domain/
|
|
||||||
internal/database/controller/
|
|
||||||
internal/database/adapter/postgresql/
|
|
||||||
internal/database/adapter/openbao/
|
|
||||||
internal/database/adapter/externalsecrets/
|
|
||||||
docs/database/
|
|
||||||
```
|
|
||||||
|
|
||||||
最终目录可按 Kubebuilder 与现有模块约定微调，但 Database 不依赖 execution/Job 模块，也不把
|
|
||||||
PostgreSQL、OpenBao 或 External Secrets 客户端放入共享万能 service/repository 层。
|
|
||||||
|
|
||||||
Database API 直接重构为 Ayatori 统一结构：API group 使用
|
|
||||||
`database.ayatori.ddupan.top/v1alpha1`，Go package 使用 `api/database/v1alpha1`，controller、
|
|
||||||
domain 与 adapter 放入 Ayatori 对应 Database 模块。原 `database.ddupan.top/v1alpha1` 不保留
|
|
||||||
别名、conversion 或兼容入口。
|
|
||||||
|
|
||||||
迁移前逐项核对批准规格、领域模型与当前 Go types；冲突时以批准规格为准。代码质量通过重写
|
|
||||||
旧运行链路、清晰 application/adapter 边界和测试实现，不通过改变已批准行为获得。无需实现在线
|
|
||||||
CRD conversion 或数据迁移。
|
|
||||||
|
|
||||||
## 迁移方式
|
|
||||||
|
|
||||||
1. 以包含已合并 Instance 领域基础和 CI #14 的最新 `main` commit 作为 source reference；记录
|
|
||||||
commit，将完整批准规格与设计文档迁入 Ayatori Database 文档，并迁移领域模型和纯单元测试。
|
|
||||||
设计合同直接复用；旧运行代码不逐文件复制。
|
|
||||||
2. 保留源仓库暂停中的脏工作树，不移动、提交或复制两个 extension observation 文件。以后可以
|
|
||||||
先在源仓库形成独立 commit，或在 Ayatori 根据批准合同重新实现，但不得把未提交内容描述为来源。
|
|
||||||
3. 在 Ayatori multi-group 项目中用 Kubebuilder 注册 Database API，按批准规格迁移 types，重新
|
|
||||||
生成 `database.ayatori.ddupan.top` CRD、DeepCopy 与 RBAC；不直接复制旧生成文件或旧 `PROJECT`。
|
|
||||||
4. 删除旧运行链路假设，以 Ayatori 当前 Go、Kubernetes 与 controller-runtime 版本重新建立
|
|
||||||
application ports 和 adapter contract；先恢复 PostgreSQL registry/adapter contract tests。
|
|
||||||
5. 逐片实现 Instance observe、Kubernetes Secret 管理凭据与连接刷新、Tenant provisioning、
|
|
||||||
OpenBao、ExternalSecret、删除与恢复流程；
|
|
||||||
每片必须包含对应单元、envtest 和真实 PostgreSQL/OpenBao 集成测试。
|
|
||||||
6. Ayatori 中的 Database 模块达到原项目验收标准并完成迁移演练后，冻结旧仓库并将其 README
|
|
||||||
指向 Ayatori；不同时运行两个 controller 管理同一组 CR。
|
|
||||||
|
|
||||||
不通过一次性 unrelated-history merge 或整仓复制保留表面上的 Git 历史。旧仓库和 source commit
|
|
||||||
保留完整来源历史；Ayatori 迁移提交按可审阅行为切片记录 provenance。
|
|
||||||
|
|
||||||
## 结果
|
|
||||||
|
|
||||||
- Ayatori 获得第一个真实产品领域，而不是继续围绕实验性 Job 扩张。
|
|
||||||
- 已批准的 DBaaS 设计与测试投资得到保留。
|
|
||||||
- 单一 manager/release 不意味着领域耦合；Database 仍保持独立 package、adapter 和测试边界。
|
|
||||||
- 可以从已合并的领域基础开始迁移；旧运行链路和未提交 extension observation 不进入首个切片。
|
|
||||||
- 无部署兼容负担允许彻底重写旧运行链路，不为尚未使用的实现技术债保留兼容层；已批准设计合同
|
|
||||||
仍然有效。
|
|
||||||
- Database 使用 Ayatori 统一 API group 与目录结构，不为未投入使用的旧 group 保留入口。
|
|
||||||
+3
-4
@@ -34,7 +34,7 @@
|
|||||||
|
|
||||||
- 建立稳定的 `VirtualMachine` 北向 API，并支持现有资源 adopt。
|
- 建立稳定的 `VirtualMachine` 北向 API，并支持现有资源 adopt。
|
||||||
- 南向按能力组合 Proxmox API、节点受限 Agent/CLI 与 `ManualTask`，不假设 Proxmox API 完整。
|
- 南向按能力组合 Proxmox API、节点受限 Agent/CLI 与 `ManualTask`，不假设 Proxmox API 完整。
|
||||||
- Node 加入、drain 和 `SafeToRemove`；Node API 由 Ayatori Compute Agent 实现，不依赖 kubelet。
|
- ComputeNode 加入、drain 和 `SafeToRemove`。
|
||||||
- StorageClass、StoragePool、Volume 与迁移计划。
|
- StorageClass、StoragePool、Volume 与迁移计划。
|
||||||
- 先支持人工磁盘迁移，再按实际收益自动化。
|
- 先支持人工磁盘迁移，再按实际收益自动化。
|
||||||
|
|
||||||
@@ -51,10 +51,9 @@
|
|||||||
- 用户集群只暴露 worker node，控制面完全由平台托管。
|
- 用户集群只暴露 worker node，控制面完全由平台托管。
|
||||||
- 本节记录候选实现边界，不构成路线图承诺。
|
- 本节记录候选实现边界，不构成路线图承诺。
|
||||||
|
|
||||||
## 后续 Compute 验收场景
|
## 首个业务里程碑
|
||||||
|
|
||||||
Database 等首批资源优先落地。Compute 开始实施后，以 Laptop Rebuild Readiness 验证节点
|
完成 Laptop Rebuild Readiness：
|
||||||
生命周期与恢复能力；该场景不作为首批 Database、LoadBalancer 或 Bucket 的交付前置条件：
|
|
||||||
|
|
||||||
1. 临时节点加入。
|
1. 临时节点加入。
|
||||||
2. laptop 上的 workload 被重建、迁移或形成可执行人工任务。
|
2. laptop 上的 workload 被重建、迁移或形成可执行人工任务。
|
||||||
|
|||||||
@@ -0,0 +1,132 @@
|
|||||||
|
package kubernetes
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
|
||||||
|
batchv1 "k8s.io/api/batch/v1"
|
||||||
|
corev1 "k8s.io/api/core/v1"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
ControllerName = "execution.ayatori.ddupan.top/kubernetes"
|
||||||
|
ReferenceType = "Job"
|
||||||
|
JobUIDLabel = "execution.ayatori.ddupan.top/job-uid"
|
||||||
|
)
|
||||||
|
|
||||||
|
var ayatoriJobGVK = schema.GroupVersionKind{
|
||||||
|
Group: executionv1alpha1.GroupVersion.Group,
|
||||||
|
Version: executionv1alpha1.GroupVersion.Version,
|
||||||
|
Kind: "Job",
|
||||||
|
}
|
||||||
|
|
||||||
|
// BuildJob translates the stable execution API into the Kubernetes adapter's
|
||||||
|
// backend object. It intentionally does not accept or expose a PodSpec.
|
||||||
|
func BuildJob(
|
||||||
|
job *executionv1alpha1.Job,
|
||||||
|
parameters *executionv1alpha1.KubernetesExecutionParameters,
|
||||||
|
resources executionv1alpha1.ExecutionResourceRequirements,
|
||||||
|
) *batchv1.Job {
|
||||||
|
backoffLimit := int32(0)
|
||||||
|
controller := true
|
||||||
|
blockOwnerDeletion := true
|
||||||
|
|
||||||
|
//nolint:modernize // ObjectMeta is promoted through embedded TypeMeta; embedlit produces invalid Go here.
|
||||||
|
return &batchv1.Job{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{
|
||||||
|
Name: job.Name,
|
||||||
|
Namespace: job.Namespace,
|
||||||
|
Labels: map[string]string{
|
||||||
|
JobUIDLabel: string(job.UID),
|
||||||
|
},
|
||||||
|
OwnerReferences: []metav1.OwnerReference{{
|
||||||
|
APIVersion: ayatoriJobGVK.GroupVersion().String(),
|
||||||
|
Kind: ayatoriJobGVK.Kind,
|
||||||
|
Name: job.Name,
|
||||||
|
UID: job.UID,
|
||||||
|
Controller: &controller,
|
||||||
|
BlockOwnerDeletion: &blockOwnerDeletion,
|
||||||
|
}},
|
||||||
|
},
|
||||||
|
Spec: batchv1.JobSpec{
|
||||||
|
BackoffLimit: &backoffLimit,
|
||||||
|
Template: corev1.PodTemplateSpec{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{JobUIDLabel: string(job.UID)}},
|
||||||
|
Spec: corev1.PodSpec{
|
||||||
|
RestartPolicy: corev1.RestartPolicyNever,
|
||||||
|
ServiceAccountName: parameters.Spec.ServiceAccountName,
|
||||||
|
RuntimeClassName: optionalString(parameters.Spec.RuntimeClassName),
|
||||||
|
NodeSelector: parameters.Spec.Scheduling.NodeSelector,
|
||||||
|
Tolerations: parameters.Spec.Scheduling.Tolerations,
|
||||||
|
SecurityContext: parameters.Spec.PodSecurityContext,
|
||||||
|
ImagePullSecrets: job.Spec.Task.ImagePullSecrets,
|
||||||
|
Containers: []corev1.Container{{
|
||||||
|
Name: "task",
|
||||||
|
Image: job.Spec.Task.Image,
|
||||||
|
ImagePullPolicy: parameters.Spec.ImagePullPolicy,
|
||||||
|
Command: job.Spec.Task.Command,
|
||||||
|
Args: job.Spec.Task.Args,
|
||||||
|
WorkingDir: job.Spec.Task.WorkingDir,
|
||||||
|
Env: environment(job.Spec.Task.Env),
|
||||||
|
Resources: resourceRequirements(resources),
|
||||||
|
}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func ValidateOwnership(owner *executionv1alpha1.Job, backend *batchv1.Job) error {
|
||||||
|
if backend.Labels[JobUIDLabel] != string(owner.UID) {
|
||||||
|
return fmt.Errorf("backend Job %s/%s is not owned by Ayatori Job UID %s", backend.Namespace, backend.Name, owner.UID)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func environment(values []executionv1alpha1.EnvVar) []corev1.EnvVar {
|
||||||
|
result := make([]corev1.EnvVar, 0, len(values))
|
||||||
|
for _, value := range values {
|
||||||
|
env := corev1.EnvVar{Name: value.Name}
|
||||||
|
if value.Value != nil {
|
||||||
|
env.Value = *value.Value
|
||||||
|
}
|
||||||
|
if value.ValueFrom != nil {
|
||||||
|
env.ValueFrom = &corev1.EnvVarSource{
|
||||||
|
SecretKeyRef: value.ValueFrom.SecretKeyRef,
|
||||||
|
ConfigMapKeyRef: value.ValueFrom.ConfigMapKeyRef,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
result = append(result, env)
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
func resourceRequirements(resources executionv1alpha1.ExecutionResourceRequirements) corev1.ResourceRequirements {
|
||||||
|
return corev1.ResourceRequirements{
|
||||||
|
Requests: resourceList(resources.Requests),
|
||||||
|
Limits: resourceList(resources.Limits),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func resourceList(values executionv1alpha1.ResourceValues) corev1.ResourceList {
|
||||||
|
result := corev1.ResourceList{}
|
||||||
|
if values.CPU != nil {
|
||||||
|
result[corev1.ResourceCPU] = values.CPU.DeepCopy()
|
||||||
|
}
|
||||||
|
if values.Memory != nil {
|
||||||
|
result[corev1.ResourceMemory] = values.Memory.DeepCopy()
|
||||||
|
}
|
||||||
|
if len(result) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
func optionalString(value string) *string {
|
||||||
|
if value == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return &value
|
||||||
|
}
|
||||||
@@ -0,0 +1,71 @@
|
|||||||
|
package kubernetes
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
|
||||||
|
corev1 "k8s.io/api/core/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/api/resource"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/types"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestBuildJob(t *testing.T) {
|
||||||
|
literal := "world"
|
||||||
|
cpuRequest := resource.MustParse("100m")
|
||||||
|
memoryLimit := resource.MustParse("128Mi")
|
||||||
|
//nolint:modernize // ObjectMeta is promoted through embedded TypeMeta; embedlit produces invalid Go here.
|
||||||
|
job := &executionv1alpha1.Job{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: "hello", Namespace: "ci", UID: types.UID("job-uid")},
|
||||||
|
Spec: executionv1alpha1.JobSpec{Task: executionv1alpha1.TaskSpec{
|
||||||
|
Image: "alpine:3.22", Command: []string{"echo"}, Args: []string{"hello"},
|
||||||
|
Env: []executionv1alpha1.EnvVar{
|
||||||
|
{Name: "TARGET", Value: &literal},
|
||||||
|
{Name: "TOKEN", ValueFrom: &executionv1alpha1.EnvVarSource{
|
||||||
|
//nolint:modernize // LocalObjectReference is an embedded Kubernetes API field.
|
||||||
|
SecretKeyRef: &corev1.SecretKeySelector{LocalObjectReference: corev1.LocalObjectReference{Name: "token"}, Key: "value"},
|
||||||
|
}},
|
||||||
|
},
|
||||||
|
}},
|
||||||
|
}
|
||||||
|
parameters := &executionv1alpha1.KubernetesExecutionParameters{Spec: executionv1alpha1.KubernetesExecutionParametersSpec{
|
||||||
|
ServiceAccountName: "runner", RuntimeClassName: "runc", ImagePullPolicy: corev1.PullIfNotPresent,
|
||||||
|
Scheduling: executionv1alpha1.KubernetesSchedulingParameters{NodeSelector: map[string]string{"role": "execution"}},
|
||||||
|
}}
|
||||||
|
resources := executionv1alpha1.ExecutionResourceRequirements{
|
||||||
|
Requests: executionv1alpha1.ResourceValues{CPU: &cpuRequest},
|
||||||
|
Limits: executionv1alpha1.ResourceValues{Memory: &memoryLimit},
|
||||||
|
}
|
||||||
|
|
||||||
|
backend := BuildJob(job, parameters, resources)
|
||||||
|
pod := backend.Spec.Template.Spec
|
||||||
|
if backend.Spec.BackoffLimit == nil || *backend.Spec.BackoffLimit != 0 {
|
||||||
|
t.Fatalf("backoffLimit = %v, want 0", backend.Spec.BackoffLimit)
|
||||||
|
}
|
||||||
|
if pod.RestartPolicy != corev1.RestartPolicyNever || pod.ServiceAccountName != "runner" {
|
||||||
|
t.Fatalf("unexpected pod execution policy: %#v", pod)
|
||||||
|
}
|
||||||
|
if pod.RuntimeClassName == nil || *pod.RuntimeClassName != "runc" {
|
||||||
|
t.Fatalf("runtimeClassName = %v, want runc", pod.RuntimeClassName)
|
||||||
|
}
|
||||||
|
container := pod.Containers[0]
|
||||||
|
if container.Resources.Requests.Cpu().Cmp(cpuRequest) != 0 || container.Resources.Limits.Memory().Cmp(memoryLimit) != 0 {
|
||||||
|
t.Fatalf("resources were not mapped: %#v", container.Resources)
|
||||||
|
}
|
||||||
|
if container.Env[1].ValueFrom == nil || container.Env[1].ValueFrom.SecretKeyRef.Name != "token" {
|
||||||
|
t.Fatalf("secret reference was not preserved: %#v", container.Env[1])
|
||||||
|
}
|
||||||
|
if backend.Labels[JobUIDLabel] != "job-uid" || backend.OwnerReferences[0].UID != job.UID {
|
||||||
|
t.Fatalf("ownership identity was not preserved: %#v", backend.ObjectMeta)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestValidateOwnership(t *testing.T) {
|
||||||
|
job := &executionv1alpha1.Job{}
|
||||||
|
job.UID = types.UID("expected")
|
||||||
|
backend := BuildJob(job, &executionv1alpha1.KubernetesExecutionParameters{}, executionv1alpha1.ExecutionResourceRequirements{})
|
||||||
|
backend.Labels[JobUIDLabel] = "different"
|
||||||
|
if err := ValidateOwnership(job, backend); err == nil {
|
||||||
|
t.Fatal("ValidateOwnership() succeeded for a different Job UID")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,397 @@
|
|||||||
|
package controller
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"slices"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
|
||||||
|
kubernetesadapter "git.ddupan.top/panxiao81/ayatori/internal/adapter/kubernetes"
|
||||||
|
batchv1 "k8s.io/api/batch/v1"
|
||||||
|
corev1 "k8s.io/api/core/v1"
|
||||||
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||||
|
"k8s.io/apimachinery/pkg/api/meta"
|
||||||
|
"k8s.io/apimachinery/pkg/api/resource"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/labels"
|
||||||
|
"k8s.io/apimachinery/pkg/types"
|
||||||
|
ctrl "sigs.k8s.io/controller-runtime"
|
||||||
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||||
|
"sigs.k8s.io/controller-runtime/pkg/log"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
jobFinalizer = "execution.ayatori.ddupan.top/job-cleanup"
|
||||||
|
reasonResultUnknown = "ResultUnknown"
|
||||||
|
)
|
||||||
|
|
||||||
|
// JobReconciler executes Ayatori Jobs using supported adapters.
|
||||||
|
type JobReconciler struct {
|
||||||
|
client.Client
|
||||||
|
Now func() time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// +kubebuilder:rbac:groups=execution.ayatori.ddupan.top,resources=jobs,verbs=get;list;watch;update;patch
|
||||||
|
// +kubebuilder:rbac:groups=execution.ayatori.ddupan.top,resources=jobs/status,verbs=get;update;patch
|
||||||
|
// +kubebuilder:rbac:groups=execution.ayatori.ddupan.top,resources=jobs/finalizers,verbs=update
|
||||||
|
// +kubebuilder:rbac:groups=execution.ayatori.ddupan.top,resources=jobclasses;kubernetesexecutionparameters,verbs=get;list;watch
|
||||||
|
// +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;delete
|
||||||
|
// +kubebuilder:rbac:groups="",resources=namespaces;serviceaccounts,verbs=get;list;watch
|
||||||
|
|
||||||
|
func (r *JobReconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.Result, error) {
|
||||||
|
logger := log.FromContext(ctx)
|
||||||
|
job := &executionv1alpha1.Job{}
|
||||||
|
if err := r.Get(ctx, request.NamespacedName, job); err != nil {
|
||||||
|
return ctrl.Result{}, client.IgnoreNotFound(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !job.DeletionTimestamp.IsZero() {
|
||||||
|
return ctrl.Result{}, r.finalize(ctx, job)
|
||||||
|
}
|
||||||
|
if isTerminal(job) {
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
if job.Spec.DesiredState == executionv1alpha1.JobDesiredStateCancelled {
|
||||||
|
return ctrl.Result{}, r.cancel(ctx, job)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !containsString(job.Finalizers, jobFinalizer) {
|
||||||
|
job.Finalizers = append(job.Finalizers, jobFinalizer)
|
||||||
|
if err := r.Update(ctx, job); err != nil {
|
||||||
|
return ctrl.Result{}, err
|
||||||
|
}
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
if job.Status.Execution != nil {
|
||||||
|
return ctrl.Result{}, r.observeExisting(ctx, job)
|
||||||
|
}
|
||||||
|
|
||||||
|
class, parameters, resources, waiting, err := r.resolve(ctx, job)
|
||||||
|
if err != nil {
|
||||||
|
return ctrl.Result{}, err
|
||||||
|
}
|
||||||
|
if waiting {
|
||||||
|
return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
backend := &batchv1.Job{}
|
||||||
|
key := types.NamespacedName{Namespace: job.Namespace, Name: job.Name}
|
||||||
|
err = r.Get(ctx, key, backend)
|
||||||
|
if apierrors.IsNotFound(err) {
|
||||||
|
backend = kubernetesadapter.BuildJob(job, parameters, resources)
|
||||||
|
if err := r.Create(ctx, backend); err != nil {
|
||||||
|
return ctrl.Result{}, err
|
||||||
|
}
|
||||||
|
logger.Info("Created Kubernetes backend Job", "backend", key)
|
||||||
|
return ctrl.Result{}, r.markScheduled(ctx, job, class, parameters, resources, backend)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return ctrl.Result{}, err
|
||||||
|
}
|
||||||
|
if err := kubernetesadapter.ValidateOwnership(job, backend); err != nil {
|
||||||
|
return ctrl.Result{}, r.setCondition(ctx, job, metav1.Condition{
|
||||||
|
Type: executionv1alpha1.JobConditionScheduled, Status: metav1.ConditionFalse,
|
||||||
|
Reason: "BackendConflict", Message: err.Error(),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
if !conditionTrue(job.Status.Conditions, executionv1alpha1.JobConditionScheduled) {
|
||||||
|
return ctrl.Result{}, r.markScheduled(ctx, job, class, parameters, resources, backend)
|
||||||
|
}
|
||||||
|
return ctrl.Result{}, r.observe(ctx, job, backend)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) observeExisting(ctx context.Context, job *executionv1alpha1.Job) error {
|
||||||
|
if job.Status.Execution.Adapter != "kubernetes" {
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{
|
||||||
|
Type: executionv1alpha1.JobConditionSucceeded, Status: metav1.ConditionUnknown,
|
||||||
|
Reason: reasonResultUnknown, Message: fmt.Sprintf("adapter %q is not available", job.Status.Execution.Adapter),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
backend := &batchv1.Job{}
|
||||||
|
key := types.NamespacedName{Namespace: job.Namespace, Name: job.Name}
|
||||||
|
if err := r.Get(ctx, key, backend); err != nil {
|
||||||
|
if apierrors.IsNotFound(err) {
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{
|
||||||
|
Type: executionv1alpha1.JobConditionSucceeded, Status: metav1.ConditionUnknown,
|
||||||
|
Reason: reasonResultUnknown, Message: "Kubernetes backend Job is missing",
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if err := kubernetesadapter.ValidateOwnership(job, backend); err != nil {
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{
|
||||||
|
Type: executionv1alpha1.JobConditionSucceeded, Status: metav1.ConditionUnknown,
|
||||||
|
Reason: reasonResultUnknown, Message: err.Error(),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return r.observe(ctx, job, backend)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) resolve(
|
||||||
|
ctx context.Context,
|
||||||
|
job *executionv1alpha1.Job,
|
||||||
|
) (*executionv1alpha1.JobClass, *executionv1alpha1.KubernetesExecutionParameters, executionv1alpha1.ExecutionResourceRequirements, bool, error) {
|
||||||
|
if job.Spec.JobClassName == "" {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, true, r.reject(ctx, job, "NoDefaultJobClass", "spec.jobClassName is required in the first implementation slice")
|
||||||
|
}
|
||||||
|
|
||||||
|
class := &executionv1alpha1.JobClass{}
|
||||||
|
if err := r.Get(ctx, types.NamespacedName{Name: job.Spec.JobClassName}, class); err != nil {
|
||||||
|
if apierrors.IsNotFound(err) {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, true, r.reject(ctx, job, "JobClassNotFound", fmt.Sprintf("JobClass %q does not exist", job.Spec.JobClassName))
|
||||||
|
}
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, false, err
|
||||||
|
}
|
||||||
|
if class.Spec.ControllerName != kubernetesadapter.ControllerName {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, true, r.reject(ctx, job, "UnsupportedController", fmt.Sprintf("controller %q is not supported", class.Spec.ControllerName))
|
||||||
|
}
|
||||||
|
ref := class.Spec.ParametersRef
|
||||||
|
if ref.Group != executionv1alpha1.GroupVersion.Group || ref.Kind != "KubernetesExecutionParameters" {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, true, r.reject(ctx, job, "InvalidParametersReference", "JobClass must reference KubernetesExecutionParameters")
|
||||||
|
}
|
||||||
|
if allowed, err := r.namespaceAllowed(ctx, job.Namespace, class.Spec.AllowedNamespaces); err != nil {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, false, err
|
||||||
|
} else if !allowed {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, true, r.reject(ctx, job, "NamespaceNotAllowed", fmt.Sprintf("namespace %q is not allowed by JobClass %q", job.Namespace, class.Name))
|
||||||
|
}
|
||||||
|
|
||||||
|
parameters := &executionv1alpha1.KubernetesExecutionParameters{}
|
||||||
|
if err := r.Get(ctx, types.NamespacedName{Name: ref.Name}, parameters); err != nil {
|
||||||
|
if apierrors.IsNotFound(err) {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, true, r.reject(ctx, job, "ParametersNotFound", fmt.Sprintf("KubernetesExecutionParameters %q does not exist", ref.Name))
|
||||||
|
}
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, false, err
|
||||||
|
}
|
||||||
|
serviceAccount := &corev1.ServiceAccount{}
|
||||||
|
if err := r.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: parameters.Spec.ServiceAccountName}, serviceAccount); err != nil {
|
||||||
|
if apierrors.IsNotFound(err) {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, true, r.reject(ctx, job, "ServiceAccountNotFound", fmt.Sprintf("ServiceAccount %q does not exist", parameters.Spec.ServiceAccountName))
|
||||||
|
}
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, false, err
|
||||||
|
}
|
||||||
|
|
||||||
|
resources := applyResourceDefaults(job.Spec.Resources, class.Spec.Resources.Defaults)
|
||||||
|
if err := validateResources(resources); err != nil {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, true, r.reject(ctx, job, "InvalidResources", err.Error())
|
||||||
|
}
|
||||||
|
if err := r.accept(ctx, job, class, parameters, resources); err != nil {
|
||||||
|
return nil, nil, executionv1alpha1.ExecutionResourceRequirements{}, false, err
|
||||||
|
}
|
||||||
|
return class, parameters, resources, false, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) namespaceAllowed(ctx context.Context, namespace string, selector *metav1.LabelSelector) (bool, error) {
|
||||||
|
if selector == nil {
|
||||||
|
return true, nil
|
||||||
|
}
|
||||||
|
ns := &corev1.Namespace{}
|
||||||
|
if err := r.Get(ctx, types.NamespacedName{Name: namespace}, ns); err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
compiled, err := metav1.LabelSelectorAsSelector(selector)
|
||||||
|
if err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
return compiled.Matches(labels.Set(ns.Labels)), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) accept(ctx context.Context, job *executionv1alpha1.Job, class *executionv1alpha1.JobClass, parameters *executionv1alpha1.KubernetesExecutionParameters, resources executionv1alpha1.ExecutionResourceRequirements) error {
|
||||||
|
job.Status.ResolvedJobClass = &executionv1alpha1.ResolvedJobClassReference{
|
||||||
|
Name: class.Name, UID: class.UID, ControllerName: class.Spec.ControllerName,
|
||||||
|
ParametersRef: executionv1alpha1.ParametersReference{
|
||||||
|
Group: class.Spec.ParametersRef.Group, Kind: class.Spec.ParametersRef.Kind,
|
||||||
|
Name: parameters.Name, UID: parameters.UID,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
job.Status.EffectiveResources = resources
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{
|
||||||
|
Type: executionv1alpha1.JobConditionAccepted, Status: metav1.ConditionTrue,
|
||||||
|
Reason: "Accepted", Message: fmt.Sprintf("JobClass %q accepted", class.Name),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) reject(ctx context.Context, job *executionv1alpha1.Job, reason, message string) error {
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{
|
||||||
|
Type: executionv1alpha1.JobConditionAccepted, Status: metav1.ConditionFalse,
|
||||||
|
Reason: reason, Message: message,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) markScheduled(ctx context.Context, job *executionv1alpha1.Job, class *executionv1alpha1.JobClass, parameters *executionv1alpha1.KubernetesExecutionParameters, resources executionv1alpha1.ExecutionResourceRequirements, backend *batchv1.Job) error {
|
||||||
|
job.Status.ResolvedJobClass = &executionv1alpha1.ResolvedJobClassReference{
|
||||||
|
Name: class.Name, UID: class.UID, ControllerName: class.Spec.ControllerName,
|
||||||
|
ParametersRef: executionv1alpha1.ParametersReference{Group: class.Spec.ParametersRef.Group, Kind: class.Spec.ParametersRef.Kind, Name: parameters.Name, UID: parameters.UID},
|
||||||
|
}
|
||||||
|
job.Status.EffectiveResources = resources
|
||||||
|
job.Status.Execution = &executionv1alpha1.ExecutionStatus{
|
||||||
|
Adapter: "kubernetes",
|
||||||
|
References: []executionv1alpha1.ExecutionReference{{Type: kubernetesadapter.ReferenceType, ID: string(backend.UID)}},
|
||||||
|
}
|
||||||
|
meta.SetStatusCondition(&job.Status.Conditions, condition(job, executionv1alpha1.JobConditionAccepted, metav1.ConditionTrue, "Accepted", "Job accepted"))
|
||||||
|
meta.SetStatusCondition(&job.Status.Conditions, condition(job, executionv1alpha1.JobConditionScheduled, metav1.ConditionTrue, "BackendCreated", "Kubernetes Job created"))
|
||||||
|
meta.SetStatusCondition(&job.Status.Conditions, condition(job, executionv1alpha1.JobConditionSucceeded, metav1.ConditionUnknown, "Pending", "Waiting for task to start"))
|
||||||
|
job.Status.ObservedGeneration = job.Generation
|
||||||
|
return r.Status().Update(ctx, job)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) observe(ctx context.Context, job *executionv1alpha1.Job, backend *batchv1.Job) error {
|
||||||
|
if job.Status.StartTime == nil && backend.Status.StartTime != nil {
|
||||||
|
job.Status.StartTime = backend.Status.StartTime.DeepCopy()
|
||||||
|
}
|
||||||
|
for _, backendCondition := range backend.Status.Conditions {
|
||||||
|
switch {
|
||||||
|
case backendCondition.Type == batchv1.JobComplete && backendCondition.Status == corev1.ConditionTrue:
|
||||||
|
completion := backend.Status.CompletionTime
|
||||||
|
if completion == nil {
|
||||||
|
now := metav1.NewTime(r.now())
|
||||||
|
completion = &now
|
||||||
|
}
|
||||||
|
job.Status.CompletionTime = completion.DeepCopy()
|
||||||
|
job.Status.Result = &executionv1alpha1.JobResult{Reason: "Completed"}
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{Type: executionv1alpha1.JobConditionSucceeded, Status: metav1.ConditionTrue, Reason: "Completed", Message: backendCondition.Message})
|
||||||
|
case backendCondition.Type == batchv1.JobFailed && backendCondition.Status == corev1.ConditionTrue:
|
||||||
|
completion := metav1.NewTime(r.now())
|
||||||
|
job.Status.CompletionTime = &completion
|
||||||
|
job.Status.Result = &executionv1alpha1.JobResult{Reason: "ProcessFailed"}
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{Type: executionv1alpha1.JobConditionSucceeded, Status: metav1.ConditionFalse, Reason: "ProcessFailed", Message: backendCondition.Message})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
reason := "Pending"
|
||||||
|
message := "Waiting for task to start"
|
||||||
|
if backend.Status.StartTime != nil || backend.Status.Active > 0 {
|
||||||
|
reason = "Running"
|
||||||
|
message = "Task is running"
|
||||||
|
}
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{Type: executionv1alpha1.JobConditionSucceeded, Status: metav1.ConditionUnknown, Reason: reason, Message: message})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) cancel(ctx context.Context, job *executionv1alpha1.Job) error {
|
||||||
|
backend := &batchv1.Job{}
|
||||||
|
key := types.NamespacedName{Namespace: job.Namespace, Name: job.Name}
|
||||||
|
err := r.Get(ctx, key, backend)
|
||||||
|
if err == nil {
|
||||||
|
if err := kubernetesadapter.ValidateOwnership(job, backend); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
for _, backendCondition := range backend.Status.Conditions {
|
||||||
|
if (backendCondition.Type == batchv1.JobComplete || backendCondition.Type == batchv1.JobFailed) &&
|
||||||
|
backendCondition.Status == corev1.ConditionTrue {
|
||||||
|
return r.observe(ctx, job, backend)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := r.Delete(ctx, backend, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil && !apierrors.IsNotFound(err) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if !apierrors.IsNotFound(err) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
now := metav1.NewTime(r.now())
|
||||||
|
job.Status.CompletionTime = &now
|
||||||
|
job.Status.Result = &executionv1alpha1.JobResult{Reason: "Cancelled"}
|
||||||
|
return r.setCondition(ctx, job, metav1.Condition{Type: executionv1alpha1.JobConditionSucceeded, Status: metav1.ConditionFalse, Reason: "Cancelled", Message: "Execution cancelled"})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) finalize(ctx context.Context, job *executionv1alpha1.Job) error {
|
||||||
|
if !containsString(job.Finalizers, jobFinalizer) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
backend := &batchv1.Job{}
|
||||||
|
key := types.NamespacedName{Namespace: job.Namespace, Name: job.Name}
|
||||||
|
if err := r.Get(ctx, key, backend); err == nil {
|
||||||
|
if err := kubernetesadapter.ValidateOwnership(job, backend); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if err := r.Delete(ctx, backend, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil && !apierrors.IsNotFound(err) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
} else if !apierrors.IsNotFound(err) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
job.Finalizers = removeString(job.Finalizers, jobFinalizer)
|
||||||
|
return r.Update(ctx, job)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) setCondition(ctx context.Context, job *executionv1alpha1.Job, next metav1.Condition) error {
|
||||||
|
meta.SetStatusCondition(&job.Status.Conditions, condition(job, next.Type, next.Status, next.Reason, next.Message))
|
||||||
|
job.Status.ObservedGeneration = job.Generation
|
||||||
|
return r.Status().Update(ctx, job)
|
||||||
|
}
|
||||||
|
|
||||||
|
func condition(job *executionv1alpha1.Job, conditionType string, status metav1.ConditionStatus, reason, message string) metav1.Condition {
|
||||||
|
return metav1.Condition{Type: conditionType, Status: status, Reason: reason, Message: message, ObservedGeneration: job.Generation}
|
||||||
|
}
|
||||||
|
|
||||||
|
func conditionTrue(conditions []metav1.Condition, conditionType string) bool {
|
||||||
|
current := meta.FindStatusCondition(conditions, conditionType)
|
||||||
|
return current != nil && current.Status == metav1.ConditionTrue
|
||||||
|
}
|
||||||
|
|
||||||
|
func isTerminal(job *executionv1alpha1.Job) bool {
|
||||||
|
current := meta.FindStatusCondition(job.Status.Conditions, executionv1alpha1.JobConditionSucceeded)
|
||||||
|
return current != nil && (current.Status == metav1.ConditionTrue || current.Status == metav1.ConditionFalse)
|
||||||
|
}
|
||||||
|
|
||||||
|
func applyResourceDefaults(requested, defaults executionv1alpha1.ExecutionResourceRequirements) executionv1alpha1.ExecutionResourceRequirements {
|
||||||
|
result := requested.DeepCopy()
|
||||||
|
if result.Requests.CPU == nil && defaults.Requests.CPU != nil {
|
||||||
|
result.Requests.CPU = copyQuantity(defaults.Requests.CPU)
|
||||||
|
}
|
||||||
|
if result.Requests.Memory == nil && defaults.Requests.Memory != nil {
|
||||||
|
result.Requests.Memory = copyQuantity(defaults.Requests.Memory)
|
||||||
|
}
|
||||||
|
if result.Limits.CPU == nil && defaults.Limits.CPU != nil {
|
||||||
|
result.Limits.CPU = copyQuantity(defaults.Limits.CPU)
|
||||||
|
}
|
||||||
|
if result.Limits.Memory == nil && defaults.Limits.Memory != nil {
|
||||||
|
result.Limits.Memory = copyQuantity(defaults.Limits.Memory)
|
||||||
|
}
|
||||||
|
return *result
|
||||||
|
}
|
||||||
|
|
||||||
|
func copyQuantity(value *resource.Quantity) *resource.Quantity {
|
||||||
|
copy := value.DeepCopy()
|
||||||
|
return ©
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateResources(resources executionv1alpha1.ExecutionResourceRequirements) error {
|
||||||
|
if resources.Requests.CPU != nil && resources.Limits.CPU != nil && resources.Requests.CPU.Cmp(*resources.Limits.CPU) > 0 {
|
||||||
|
return fmt.Errorf("CPU request must not exceed limit")
|
||||||
|
}
|
||||||
|
if resources.Requests.Memory != nil && resources.Limits.Memory != nil && resources.Requests.Memory.Cmp(*resources.Limits.Memory) > 0 {
|
||||||
|
return fmt.Errorf("memory request must not exceed limit")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func containsString(values []string, target string) bool {
|
||||||
|
return slices.Contains(values, target)
|
||||||
|
}
|
||||||
|
|
||||||
|
func removeString(values []string, target string) []string {
|
||||||
|
result := values[:0]
|
||||||
|
for _, value := range values {
|
||||||
|
if value != target {
|
||||||
|
result = append(result, value)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) now() time.Time {
|
||||||
|
if r.Now != nil {
|
||||||
|
return r.Now()
|
||||||
|
}
|
||||||
|
return time.Now()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *JobReconciler) SetupWithManager(manager ctrl.Manager) error {
|
||||||
|
return ctrl.NewControllerManagedBy(manager).
|
||||||
|
For(&executionv1alpha1.Job{}).
|
||||||
|
Owns(&batchv1.Job{}).
|
||||||
|
Named("execution-job").
|
||||||
|
Complete(r)
|
||||||
|
}
|
||||||
@@ -0,0 +1,202 @@
|
|||||||
|
package controller
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
|
||||||
|
kubernetesadapter "git.ddupan.top/panxiao81/ayatori/internal/adapter/kubernetes"
|
||||||
|
batchv1 "k8s.io/api/batch/v1"
|
||||||
|
corev1 "k8s.io/api/core/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/api/meta"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime"
|
||||||
|
"k8s.io/apimachinery/pkg/types"
|
||||||
|
ctrl "sigs.k8s.io/controller-runtime"
|
||||||
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||||
|
"sigs.k8s.io/controller-runtime/pkg/envtest"
|
||||||
|
metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
integrationNamespace = "controller-integration"
|
||||||
|
integrationClass = "integration"
|
||||||
|
)
|
||||||
|
|
||||||
|
//nolint:modernize // Kubernetes API structs expose ObjectMeta through embedded TypeMeta fields.
|
||||||
|
func TestJobControllerIntegration(t *testing.T) {
|
||||||
|
if os.Getenv("KUBEBUILDER_ASSETS") == "" {
|
||||||
|
t.Skip("KUBEBUILDER_ASSETS is unset; run make test to execute controller integration tests")
|
||||||
|
}
|
||||||
|
|
||||||
|
scheme := runtime.NewScheme()
|
||||||
|
for _, addToScheme := range []func(*runtime.Scheme) error{
|
||||||
|
corev1.AddToScheme,
|
||||||
|
batchv1.AddToScheme,
|
||||||
|
executionv1alpha1.AddToScheme,
|
||||||
|
} {
|
||||||
|
if err := addToScheme(scheme); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
crdPath, err := filepath.Abs("../../config/crd/bases")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
environment := &envtest.Environment{CRDDirectoryPaths: []string{crdPath}}
|
||||||
|
config, err := environment.Start()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("start envtest: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() {
|
||||||
|
if err := environment.Stop(); err != nil {
|
||||||
|
t.Errorf("stop envtest: %v", err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
manager, err := ctrl.NewManager(config, ctrl.Options{
|
||||||
|
Scheme: scheme,
|
||||||
|
Metrics: metricsserver.Options{BindAddress: "0"},
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := (&JobReconciler{Client: manager.GetClient()}).SetupWithManager(manager); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
managerContext, cancelManager := context.WithCancel(context.Background())
|
||||||
|
t.Cleanup(cancelManager)
|
||||||
|
managerErrors := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
managerErrors <- manager.Start(managerContext)
|
||||||
|
}()
|
||||||
|
if !manager.GetCache().WaitForCacheSync(managerContext) {
|
||||||
|
t.Fatal("manager cache did not synchronize")
|
||||||
|
}
|
||||||
|
|
||||||
|
directClient, err := client.New(config, client.Options{Scheme: scheme})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
ctx := context.Background()
|
||||||
|
objects := []client.Object{
|
||||||
|
&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: integrationNamespace, Labels: map[string]string{testLabelKey: testLabelEnabled}}},
|
||||||
|
&corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: testSAName, Namespace: integrationNamespace}},
|
||||||
|
&executionv1alpha1.KubernetesExecutionParameters{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: integrationClass},
|
||||||
|
Spec: executionv1alpha1.KubernetesExecutionParametersSpec{
|
||||||
|
ServiceAccountName: testSAName,
|
||||||
|
ImagePullPolicy: corev1.PullIfNotPresent,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
&executionv1alpha1.JobClass{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: integrationClass},
|
||||||
|
Spec: executionv1alpha1.JobClassSpec{
|
||||||
|
ControllerName: kubernetesadapter.ControllerName,
|
||||||
|
ParametersRef: executionv1alpha1.ParametersReference{
|
||||||
|
Group: executionv1alpha1.GroupVersion.Group,
|
||||||
|
Kind: "KubernetesExecutionParameters",
|
||||||
|
Name: integrationClass,
|
||||||
|
},
|
||||||
|
AllowedNamespaces: &metav1.LabelSelector{MatchLabels: map[string]string{testLabelKey: testLabelEnabled}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
for _, object := range objects {
|
||||||
|
if err := directClient.Create(ctx, object); err != nil {
|
||||||
|
t.Fatalf("create %T: %v", object, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
job := &executionv1alpha1.Job{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: testJobName, Namespace: integrationNamespace},
|
||||||
|
Spec: executionv1alpha1.JobSpec{
|
||||||
|
JobClassName: integrationClass,
|
||||||
|
DesiredState: executionv1alpha1.JobDesiredStateRunning,
|
||||||
|
Task: executionv1alpha1.TaskSpec{Image: "alpine:3.22", Command: []string{"true"}},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
if err := directClient.Create(ctx, job); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
backend := &batchv1.Job{}
|
||||||
|
eventually(t, 10*time.Second, func() (bool, error) {
|
||||||
|
err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, backend)
|
||||||
|
return err == nil, client.IgnoreNotFound(err)
|
||||||
|
})
|
||||||
|
if backend.Labels[kubernetesadapter.JobUIDLabel] != string(job.UID) {
|
||||||
|
t.Fatalf("backend identity label = %q, want %q", backend.Labels[kubernetesadapter.JobUIDLabel], job.UID)
|
||||||
|
}
|
||||||
|
|
||||||
|
eventually(t, 10*time.Second, func() (bool, error) {
|
||||||
|
if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, job); err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
return conditionStatus(job, executionv1alpha1.JobConditionScheduled) == metav1.ConditionTrue, nil
|
||||||
|
})
|
||||||
|
|
||||||
|
completed := metav1.Now()
|
||||||
|
if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, backend); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
backend.Status.StartTime = &completed
|
||||||
|
backend.Status.CompletionTime = &completed
|
||||||
|
backend.Status.Conditions = []batchv1.JobCondition{
|
||||||
|
{Type: batchv1.JobSuccessCriteriaMet, Status: corev1.ConditionTrue, Reason: "CompletionsReached"},
|
||||||
|
{Type: batchv1.JobComplete, Status: corev1.ConditionTrue, Reason: "Completed"},
|
||||||
|
}
|
||||||
|
if err := directClient.Status().Update(ctx, backend); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
eventually(t, 10*time.Second, func() (bool, error) {
|
||||||
|
if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, job); err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
return conditionStatus(job, executionv1alpha1.JobConditionSucceeded) == metav1.ConditionTrue, nil
|
||||||
|
})
|
||||||
|
if job.Status.StartTime == nil || job.Status.CompletionTime == nil || job.Status.Execution == nil {
|
||||||
|
t.Fatalf("controller did not persist execution status: %#v", job.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
cancelManager()
|
||||||
|
select {
|
||||||
|
case err := <-managerErrors:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("manager stopped with error: %v", err)
|
||||||
|
}
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("manager did not stop")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func eventually(t *testing.T, timeout time.Duration, check func() (bool, error)) {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.Now().Add(timeout)
|
||||||
|
for time.Now().Before(deadline) {
|
||||||
|
ready, err := check()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if ready {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
}
|
||||||
|
t.Fatal(fmt.Errorf("condition was not met within %s", timeout))
|
||||||
|
}
|
||||||
|
|
||||||
|
func conditionStatus(job *executionv1alpha1.Job, conditionType string) metav1.ConditionStatus {
|
||||||
|
condition := meta.FindStatusCondition(job.Status.Conditions, conditionType)
|
||||||
|
if condition == nil {
|
||||||
|
return metav1.ConditionUnknown
|
||||||
|
}
|
||||||
|
return condition.Status
|
||||||
|
}
|
||||||
@@ -0,0 +1,245 @@
|
|||||||
|
package controller
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
|
||||||
|
kubernetesadapter "git.ddupan.top/panxiao81/ayatori/internal/adapter/kubernetes"
|
||||||
|
batchv1 "k8s.io/api/batch/v1"
|
||||||
|
corev1 "k8s.io/api/core/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/api/meta"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime"
|
||||||
|
"k8s.io/apimachinery/pkg/types"
|
||||||
|
ctrl "sigs.k8s.io/controller-runtime"
|
||||||
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||||
|
"sigs.k8s.io/controller-runtime/pkg/client/fake"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
defaultClassName = "default"
|
||||||
|
testJobName = "hello"
|
||||||
|
testSAName = "runner"
|
||||||
|
testLabelKey = "execution"
|
||||||
|
testLabelEnabled = "enabled"
|
||||||
|
)
|
||||||
|
|
||||||
|
//nolint:modernize // controller-runtime and Kubernetes API structs expose promoted embedded fields.
|
||||||
|
func TestJobReconcilerKubernetesLifecycle(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
now := time.Unix(1_700_000_000, 0)
|
||||||
|
reconciler, kubeClient := testReconciler(t, now, validObjects()...)
|
||||||
|
request := ctrl.Request{}
|
||||||
|
request.NamespacedName = types.NamespacedName{Namespace: "ci", Name: testJobName}
|
||||||
|
|
||||||
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
||||||
|
t.Fatalf("add finalizer: %v", err)
|
||||||
|
}
|
||||||
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
||||||
|
t.Fatalf("create backend: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
backend := &batchv1.Job{}
|
||||||
|
if err := kubeClient.Get(ctx, request.NamespacedName, backend); err != nil {
|
||||||
|
t.Fatalf("backend Job was not created: %v", err)
|
||||||
|
}
|
||||||
|
if backend.Labels[kubernetesadapter.JobUIDLabel] != "ayatori-job-uid" {
|
||||||
|
t.Fatalf("backend UID label = %q", backend.Labels[kubernetesadapter.JobUIDLabel])
|
||||||
|
}
|
||||||
|
|
||||||
|
job := getJob(t, ctx, kubeClient, request.NamespacedName)
|
||||||
|
if !conditionIs(job, executionv1alpha1.JobConditionAccepted, metav1.ConditionTrue) ||
|
||||||
|
!conditionIs(job, executionv1alpha1.JobConditionScheduled, metav1.ConditionTrue) {
|
||||||
|
t.Fatalf("Job was not accepted and scheduled: %#v", job.Status.Conditions)
|
||||||
|
}
|
||||||
|
|
||||||
|
started := metav1.NewTime(now.Add(time.Minute))
|
||||||
|
backend.Status.StartTime = &started
|
||||||
|
backend.Status.Active = 1
|
||||||
|
if err := kubeClient.Status().Update(ctx, backend); err != nil {
|
||||||
|
t.Fatalf("set backend running: %v", err)
|
||||||
|
}
|
||||||
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
||||||
|
t.Fatalf("observe running backend: %v", err)
|
||||||
|
}
|
||||||
|
job = getJob(t, ctx, kubeClient, request.NamespacedName)
|
||||||
|
if job.Status.StartTime == nil || !conditionIs(job, executionv1alpha1.JobConditionSucceeded, metav1.ConditionUnknown) {
|
||||||
|
t.Fatalf("running state was not observed: %#v", job.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
completed := metav1.NewTime(now.Add(2 * time.Minute))
|
||||||
|
backend = &batchv1.Job{}
|
||||||
|
if err := kubeClient.Get(ctx, request.NamespacedName, backend); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
backend.Status.Active = 0
|
||||||
|
backend.Status.CompletionTime = &completed
|
||||||
|
backend.Status.Conditions = []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue, Reason: "Completed"}}
|
||||||
|
if err := kubeClient.Status().Update(ctx, backend); err != nil {
|
||||||
|
t.Fatalf("set backend complete: %v", err)
|
||||||
|
}
|
||||||
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
||||||
|
t.Fatalf("observe completed backend: %v", err)
|
||||||
|
}
|
||||||
|
job = getJob(t, ctx, kubeClient, request.NamespacedName)
|
||||||
|
if !conditionIs(job, executionv1alpha1.JobConditionSucceeded, metav1.ConditionTrue) || job.Status.CompletionTime == nil {
|
||||||
|
t.Fatalf("terminal state was not observed: %#v", job.Status)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:modernize // controller-runtime Request exposes NamespacedName as a promoted embedded field.
|
||||||
|
func TestJobReconcilerRejectsMissingClass(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
job := validObjects()[3].(*executionv1alpha1.Job).DeepCopy()
|
||||||
|
job.Spec.JobClassName = "missing"
|
||||||
|
reconciler, kubeClient := testReconciler(t, time.Now(), validObjects()[0], validObjects()[1], job)
|
||||||
|
request := ctrl.Request{}
|
||||||
|
request.NamespacedName = types.NamespacedName{Namespace: job.Namespace, Name: job.Name}
|
||||||
|
|
||||||
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
result, err := reconciler.Reconcile(ctx, request)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if result.RequeueAfter == 0 {
|
||||||
|
t.Fatal("missing JobClass did not schedule a retry")
|
||||||
|
}
|
||||||
|
stored := getJob(t, ctx, kubeClient, request.NamespacedName)
|
||||||
|
accepted := meta.FindStatusCondition(stored.Status.Conditions, executionv1alpha1.JobConditionAccepted)
|
||||||
|
if accepted == nil || accepted.Status != metav1.ConditionFalse || accepted.Reason != "JobClassNotFound" {
|
||||||
|
t.Fatalf("unexpected Accepted condition: %#v", accepted)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:modernize // controller-runtime Request exposes NamespacedName as a promoted embedded field.
|
||||||
|
func TestJobReconcilerObservesExistingExecutionWithoutJobClass(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
now := time.Unix(1_700_000_000, 0)
|
||||||
|
job := validObjects()[3].(*executionv1alpha1.Job).DeepCopy()
|
||||||
|
job.Finalizers = []string{jobFinalizer}
|
||||||
|
job.Status.Execution = &executionv1alpha1.ExecutionStatus{Adapter: "kubernetes"}
|
||||||
|
backend := kubernetesadapter.BuildJob(job, validObjects()[4].(*executionv1alpha1.KubernetesExecutionParameters), executionv1alpha1.ExecutionResourceRequirements{})
|
||||||
|
backend.Status.StartTime = &metav1.Time{Time: now}
|
||||||
|
backend.Status.Active = 1
|
||||||
|
reconciler, kubeClient := testReconciler(t, now, job, backend)
|
||||||
|
request := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: job.Namespace, Name: job.Name}}
|
||||||
|
|
||||||
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
stored := getJob(t, ctx, kubeClient, request.NamespacedName)
|
||||||
|
if stored.Status.StartTime == nil || !conditionIs(stored, executionv1alpha1.JobConditionSucceeded, metav1.ConditionUnknown) {
|
||||||
|
t.Fatalf("existing execution was not observed without its JobClass: %#v", stored.Status)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:modernize // controller-runtime Request exposes NamespacedName as a promoted embedded field.
|
||||||
|
func TestJobReconcilerCancelsBeforeScheduling(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
job := validObjects()[3].(*executionv1alpha1.Job).DeepCopy()
|
||||||
|
job.Spec.DesiredState = executionv1alpha1.JobDesiredStateCancelled
|
||||||
|
reconciler, kubeClient := testReconciler(t, time.Unix(1_700_000_000, 0), job)
|
||||||
|
request := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: job.Namespace, Name: job.Name}}
|
||||||
|
|
||||||
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
stored := getJob(t, ctx, kubeClient, request.NamespacedName)
|
||||||
|
condition := meta.FindStatusCondition(stored.Status.Conditions, executionv1alpha1.JobConditionSucceeded)
|
||||||
|
if condition == nil || condition.Status != metav1.ConditionFalse || condition.Reason != "Cancelled" {
|
||||||
|
t.Fatalf("unexpected cancellation condition: %#v", condition)
|
||||||
|
}
|
||||||
|
if stored.Status.CompletionTime == nil {
|
||||||
|
t.Fatal("cancelled Job has no completionTime")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:modernize // controller-runtime Request exposes NamespacedName as a promoted embedded field.
|
||||||
|
func TestJobReconcilerKeepsConfirmedSuccessDuringCancellation(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
now := time.Unix(1_700_000_000, 0)
|
||||||
|
job := validObjects()[3].(*executionv1alpha1.Job).DeepCopy()
|
||||||
|
job.Spec.DesiredState = executionv1alpha1.JobDesiredStateCancelled
|
||||||
|
job.Finalizers = []string{jobFinalizer}
|
||||||
|
backend := kubernetesadapter.BuildJob(job, validObjects()[4].(*executionv1alpha1.KubernetesExecutionParameters), executionv1alpha1.ExecutionResourceRequirements{})
|
||||||
|
backend.Status.CompletionTime = &metav1.Time{Time: now}
|
||||||
|
backend.Status.Conditions = []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue}}
|
||||||
|
reconciler, kubeClient := testReconciler(t, now, job, backend)
|
||||||
|
request := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: job.Namespace, Name: job.Name}}
|
||||||
|
|
||||||
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
stored := getJob(t, ctx, kubeClient, request.NamespacedName)
|
||||||
|
if !conditionIs(stored, executionv1alpha1.JobConditionSucceeded, metav1.ConditionTrue) {
|
||||||
|
t.Fatalf("confirmed success was overwritten by cancellation: %#v", stored.Status.Conditions)
|
||||||
|
}
|
||||||
|
if err := kubeClient.Get(ctx, request.NamespacedName, &batchv1.Job{}); err != nil {
|
||||||
|
t.Fatalf("successful backend was deleted: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:modernize // Kubernetes API structs expose ObjectMeta through embedded TypeMeta fields.
|
||||||
|
func validObjects() []client.Object {
|
||||||
|
return []client.Object{
|
||||||
|
&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ci", Labels: map[string]string{testLabelKey: testLabelEnabled}}},
|
||||||
|
&corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: testSAName, Namespace: "ci"}},
|
||||||
|
&executionv1alpha1.JobClass{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: defaultClassName, UID: types.UID("class-uid")},
|
||||||
|
Spec: executionv1alpha1.JobClassSpec{
|
||||||
|
ControllerName: kubernetesadapter.ControllerName,
|
||||||
|
ParametersRef: executionv1alpha1.ParametersReference{Group: executionv1alpha1.GroupVersion.Group, Kind: "KubernetesExecutionParameters", Name: defaultClassName},
|
||||||
|
AllowedNamespaces: &metav1.LabelSelector{MatchLabels: map[string]string{testLabelKey: testLabelEnabled}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
&executionv1alpha1.Job{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: testJobName, Namespace: "ci", UID: types.UID("ayatori-job-uid")},
|
||||||
|
Spec: executionv1alpha1.JobSpec{
|
||||||
|
JobClassName: defaultClassName, DesiredState: executionv1alpha1.JobDesiredStateRunning,
|
||||||
|
Task: executionv1alpha1.TaskSpec{Image: "alpine:3.22", Command: []string{"true"}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
&executionv1alpha1.KubernetesExecutionParameters{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: defaultClassName, UID: types.UID("parameters-uid")},
|
||||||
|
Spec: executionv1alpha1.KubernetesExecutionParametersSpec{ServiceAccountName: testSAName, ImagePullPolicy: corev1.PullIfNotPresent},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func testReconciler(t *testing.T, now time.Time, objects ...client.Object) (*JobReconciler, client.Client) {
|
||||||
|
t.Helper()
|
||||||
|
scheme := runtime.NewScheme()
|
||||||
|
if err := corev1.AddToScheme(scheme); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := batchv1.AddToScheme(scheme); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := executionv1alpha1.AddToScheme(scheme); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
kubeClient := fake.NewClientBuilder().
|
||||||
|
WithScheme(scheme).
|
||||||
|
WithStatusSubresource(&executionv1alpha1.Job{}, &batchv1.Job{}).
|
||||||
|
WithObjects(objects...).
|
||||||
|
Build()
|
||||||
|
return &JobReconciler{Client: kubeClient, Now: func() time.Time { return now }}, kubeClient
|
||||||
|
}
|
||||||
|
|
||||||
|
func getJob(t *testing.T, ctx context.Context, kubeClient client.Client, key types.NamespacedName) *executionv1alpha1.Job {
|
||||||
|
t.Helper()
|
||||||
|
job := &executionv1alpha1.Job{}
|
||||||
|
if err := kubeClient.Get(ctx, key, job); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return job
|
||||||
|
}
|
||||||
|
|
||||||
|
func conditionIs(job *executionv1alpha1.Job, conditionType string, status metav1.ConditionStatus) bool {
|
||||||
|
condition := meta.FindStatusCondition(job.Status.Conditions, conditionType)
|
||||||
|
return condition != nil && condition.Status == status
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user