Compare commits
10
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 panxiao81
|
ff21c5bfba
|
||
|
panxiao81
|
c004aca1cc
|
||
|
panxiao81
|
85b61a6a23
|
||
|
panxiao81
|
4206cfbab7
|
||
|
panxiao81
|
09a0a5665f
|
||
|
panxiao81
|
d9d99b2f9b
|
||
|
panxiao81
|
3e21de5942
|
||
|
panxiao81
|
f76978ca17
|
||
|
panxiao81
|
06bc54e3cf
|
||
|
panxiao81
|
86effb72a8
|
@@ -3,11 +3,47 @@
|
||||
- 本仓库是 ddupan.top homelab 的内部基础设施控制平面，不以通用发行版为初期目标。
|
||||
- 提交、文档和代码注释优先使用中文；公共 API 标识符和代码遵循对应语言惯例。
|
||||
- 不要重新实现已有成熟后端的核心能力；新增实现前先确认能否通过稳定 API 进行薄适配。
|
||||
- 不要按传统私有云或公有云产品清单推导 Ayatori 应实现的资源。新增北向 API 前必须证明 homelab
|
||||
存在真实、重复的管理缺口，现有成熟 API/IaC 不能提供足够的生命周期、状态或权限体验；“后端
|
||||
能做到”或“其他云平台提供”本身不是产品需求。
|
||||
- 当前已确认的首要产品方向是 Database、LoadBalancer、Bucket/Object Storage；VirtualMachine
|
||||
也具有明确价值，但南向实现较重。Run/Job 是验证 controller 与 adapter 的内部执行切片，不应
|
||||
自动演化为 FaaS、Cloud Run 或应用托管产品。KaaS 仅在出现真实需求时评估，不是必达终点。
|
||||
- Ayatori 复用 Kubernetes 的核心目标是 API machinery：对象存储与并发控制、list/watch、
|
||||
informer、RBAC、admission、版本化 API 和审计；不要据此推断 Ayatori 是 Kubernetes
|
||||
workload 平台，也不要默认复用 Kubernetes 的调度与数据面语义。
|
||||
- 复用 Kubernetes 内置资源只表示采用其 API contract，不表示必须运行或模拟上游实现组件。
|
||||
例如 Ayatori Compute Agent 可以直接实现 `core/v1 Node` 与 Lease 的状态语义，Ayatori
|
||||
controller 可以自行消费 Node；不得仅因使用 Node 推导必须引入 kubelet、Pod、CRI、
|
||||
kube-scheduler 或 kube-controller-manager。对每个复用资源分别明确 producer、consumer、
|
||||
ownership 与实际采用的字段语义。
|
||||
- kube-apiserver 是 Ayatori 的 API 与状态协调平面，不是领域调度器。资源的调度、生命周期、
|
||||
故障恢复、垃圾回收和后端收敛由 Ayatori controllers 实现；新增能力前应明确其属于 API
|
||||
machinery、Ayatori 领域控制循环还是外部 backend，避免把职责放错层。
|
||||
- Kubernetes、OpenSandbox、Proxmox 等均是 Ayatori 的可替换 backend/executor。除管理组件自身
|
||||
的部署外，不得仅因 controller 运行在 Kubernetes 中，就把原生 Pod、Job、Service、
|
||||
NetworkPolicy、owner reference 或同 namespace 行为作为领域 API 的隐含语义；需要这些能力时
|
||||
必须由 adapter 契约显式表达，并考虑后端位于其他集群或完全不是 Kubernetes 的情况。
|
||||
- 不要以减少自有 controller 数量为目的引入 generic-apiserver、聚合 API Server 或自行实现
|
||||
API Server。只有 CRD/kube-apiserver 在存储、API 语义或扩展能力上形成已验证的阻碍时，才评估
|
||||
接管 watch、RBAC、版本兼容和存储迁移等复杂度；controller 工作本身不会因此消失。
|
||||
- 在自行设计通用控制循环、资源生命周期、调度、回收或故障恢复机制前，先调查 Kubernetes
|
||||
核心及成熟开源 controller/operator 的实现；优先复用经过验证的模式，并记录有意偏离的
|
||||
理由。
|
||||
- 不要引入统一包装所有能力的 Application CRD；应用应直接组合正交的平台资源。
|
||||
- Proxmox VM 的北向管理不能假定单一 API 覆盖完整生命周期。允许按能力组合 Proxmox API、节点
|
||||
上的受限强类型 Agent/CLI 操作和 ManualTask；节点 Agent 不得退化为无版本契约的任意远程 shell。
|
||||
- 所有 controller 必须考虑幂等、observe、finalizer、conditions、删除策略和恢复行为。
|
||||
- Ayatori 会联动 Kubernetes API、虚拟化、存储、网络及其他外部控制面；集成测试是功能完成
|
||||
标准的一部分，不得仅凭 fake client 或 mock 测试宣告 controller、adapter 或生命周期变更完成。
|
||||
- 测试应按风险分层：纯领域规则使用快速单元测试；API schema、CEL、status subresource、
|
||||
watch/cache、owner reference 和 reconcile 事件链使用 envtest；需要 scheduler、kubelet、网络、
|
||||
存储或真实后端行为的路径在 Dev 集群或对应后端环境执行端到端测试。
|
||||
- fake client 适合穷举状态机和错误分支，但它不会完整执行 API server defaulting、validation、
|
||||
resourceVersion、garbage collection 或新版 Kubernetes 约束；涉及这些语义时必须增加真实 API
|
||||
server 测试。跨 adapter 的共同契约应使用同一套 contract tests，避免各实现产生语义漂移。
|
||||
- 集成测试必须覆盖正常路径以及幂等重试、controller 重启、依赖稍后出现、删除/finalizer、
|
||||
后端结果不确定和并发竞态等恢复路径；无法在当前层测试的部分要明确记录由哪一层验证。
|
||||
- Secret、token、kubeconfig 及具体生产凭据不得提交到仓库。
|
||||
- `deploy/dev/` 与 `deploy/prod/` 使用相同制品；生产版本只通过 promotion 更新。
|
||||
- 内部专用不构成降低测试、版本、恢复、安全和可审计要求的理由。
|
||||
|
||||
@@ -28,11 +28,15 @@ Ayatori 是 `ddupan.top` homelab 的内部基础设施控制平面。它以 Kube
|
||||
- [执行模型](docs/concepts/execution-model.md)
|
||||
- [环境与发布](docs/concepts/environments.md)
|
||||
- [路线图](docs/roadmap.md)
|
||||
- [ADR-0001：采用 Kubernetes API 作为资源模型](docs/decisions/0001-kubernetes-api-machinery.md)
|
||||
- [ADR-0001：采用 Kubernetes API machinery 作为状态协调平面](docs/decisions/0001-kubernetes-api-machinery.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-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)
|
||||
|
||||
## 当前状态
|
||||
|
||||
Ayatori 处于设计与早期实现阶段。第一个纵向切片计划是统一 Job API 与 Kubernetes
|
||||
Pod executor，随后接入 OpenSandbox executor。
|
||||
Ayatori 处于设计与早期实现阶段。当前使用 Job controller 验证第一个完整控制循环与 adapter
|
||||
边界；它不是通用 Job Service 或 FaaS 产品承诺。首批实际产品方向是 Database、LoadBalancer
|
||||
和 Bucket/Object Storage，具体顺序按纵向价值决定。
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
"sigs.k8s.io/controller-runtime/pkg/webhook"
|
||||
|
||||
executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
|
||||
"git.ddupan.top/panxiao81/ayatori/internal/controller"
|
||||
// +kubebuilder:scaffold:imports
|
||||
)
|
||||
|
||||
@@ -165,6 +166,13 @@ func main() {
|
||||
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
|
||||
|
||||
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
|
||||
|
||||
+53
-6
@@ -1,11 +1,58 @@
|
||||
---
|
||||
apiVersion: rbac.authorization.k8s.io/v1
|
||||
kind: ClusterRole
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: ayatori
|
||||
app.kubernetes.io/managed-by: kustomize
|
||||
name: manager-role
|
||||
rules:
|
||||
- apiGroups: [""]
|
||||
resources: ["pods"]
|
||||
verbs: ["get", "list", "watch"]
|
||||
- apiGroups:
|
||||
- ""
|
||||
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
|
||||
|
||||
@@ -4,9 +4,11 @@
|
||||
Git / CLI / Backstage
|
||||
│
|
||||
▼
|
||||
Kubernetes API + CRD
|
||||
kube-apiserver + etcd + CRD
|
||||
API / state coordination plane
|
||||
│
|
||||
Ayatori controllers
|
||||
Ayatori controller-manager
|
||||
scheduling / lifecycle / recovery / GC
|
||||
│
|
||||
┌──────┼──────────────┐
|
||||
│ │ │
|
||||
@@ -19,6 +21,20 @@ Terraform OpenBao / DNS / KaaS
|
||||
Ansible
|
||||
```
|
||||
|
||||
Ayatori 复用 Kubernetes 的 API machinery，而不是 Kubernetes 的容器编排产品边界。
|
||||
kube-apiserver 提供版本化对象、并发控制、list/watch、RBAC、admission 和审计；Ayatori
|
||||
controller-manager 承担所有领域控制循环。Kubernetes workload 集群只是与 OpenSandbox、
|
||||
Proxmox 等并列的 executor/backend，不默认等于运行 controller 的 management environment。
|
||||
|
||||
因此，领域 API 不得依赖“资源最终一定变成同集群原生对象”的假设。原生 Pod、Job、Service、
|
||||
NetworkPolicy、namespace 共置与 owner reference 只有在 Kubernetes adapter 内才具有原生含义；
|
||||
跨后端所需能力必须由领域模型显式定义。
|
||||
|
||||
内置 API 类型也按相同原则选择性复用。采用 `core/v1 Node` 作为计算节点 API 时，可以由
|
||||
Ayatori Compute Agent 写入状态、由 Ayatori 自有调度 controller 消费；这不会引入 kubelet、
|
||||
Pod 或 kube-scheduler。API contract、负责实现它的 controller/agent 和数据面是三个独立决策，
|
||||
不得从其中一个自动推导另外两个。
|
||||
|
||||
## 控制面
|
||||
|
||||
Dev 与 Prod 使用独立的 Kubernetes API、数据库、身份和 controller 实例。两者可以
|
||||
@@ -34,16 +50,16 @@ Ayatori 不承载或重新实现数据面。控制面故障只应阻止创建与
|
||||
|
||||
## 资源分层
|
||||
|
||||
平台提供正交产品能力，例如：
|
||||
平台只为已经验证的管理缺口提供正交产品能力。当前优先资源为：
|
||||
|
||||
- `Job`、`Sandbox`、`ManualTask`
|
||||
- `VirtualMachine`
|
||||
- `LoadBalancer`
|
||||
- `Database`
|
||||
- `Bucket`
|
||||
- `DNSRecord`
|
||||
- `Credential`
|
||||
- `KubernetesCluster`
|
||||
- `VirtualMachine`
|
||||
|
||||
`Run`/当前实验性的 `Job`、`ManualTask` 等可以作为控制面执行原语，但不是因为底层能运行 OCI
|
||||
image 就自动成为面向使用者的计算产品。`DNSRecord`、`Credential`、`KubernetesCluster` 等只在
|
||||
出现独立生命周期和真实消费者后加入；尤其 KaaS 不是预定终点。
|
||||
|
||||
只有具备独立领域生命周期的能力才应成为高阶资源。应用本身通过 GitOps 组合上述资源，
|
||||
重复组合可通过模板或 Composition 表达，而不是扩展中央 Application API。
|
||||
@@ -56,5 +72,17 @@ Ayatori 不承载或重新实现数据面。控制面故障只应阻止创建与
|
||||
2. 通过固定版本的 Terraform module 或 Ansible playbook 执行。
|
||||
3. 仅在必要时使用 GitOps bridge。
|
||||
|
||||
Proxmox 是已知例外：其远程 API 不能覆盖所需的完整 VM 生命周期。VirtualMachine adapter 可以
|
||||
按操作能力选择 Proxmox API、部署在节点上的受限强类型 Agent/CLI，或生成 `ManualTask`。Agent
|
||||
必须提供版本化操作、幂等查询、operation ID 与审计，不能暴露任意 shell，也不能把 CLI 输出
|
||||
直接当作长期稳定协议。
|
||||
|
||||
Controller 无论采用哪种执行方式，都必须提供一致的 ownership、conditions、删除语义、
|
||||
错误分类和恢复行为。
|
||||
|
||||
## API Server 边界
|
||||
|
||||
首选 kube-apiserver + CRD，持续复用其成熟的 watch、RBAC、版本化存储和 API 生态。
|
||||
generic-apiserver 或聚合 API Server 不会减少领域 controller 的数量，只会把资源服务端、
|
||||
兼容性和存储迁移责任转移给 Ayatori。只有 CRD 的限制已经形成可复现、不可通过合理领域建模
|
||||
解决的阻碍时，才重新评估自建 API Server。
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# ADR-0001：采用 Kubernetes API 作为资源模型
|
||||
# ADR-0001：采用 Kubernetes API machinery 作为状态协调平面
|
||||
|
||||
- 状态：Accepted
|
||||
- 日期：2026-09-17
|
||||
@@ -10,16 +10,49 @@ homelab 的基础设施状态分散在多套工具和后端中。仅集中 IaC
|
||||
|
||||
## 决策
|
||||
|
||||
Ayatori 使用 Kubernetes API machinery 与 CRD 表达平台资源、引用和状态，但不将平台
|
||||
限定为容器编排系统。Controller 可以运行于专用 management environment，并管理集群外
|
||||
的 VM、LB、数据库、对象存储、DNS、凭据和托管 Kubernetes 控制面。
|
||||
Ayatori 使用 kube-apiserver、etcd、Kubernetes API machinery 与 CRD 构成 API 和状态协调
|
||||
平面。主要复用的是以下难以可靠重建的能力：
|
||||
|
||||
- 版本化对象 API、schema、defaulting、validation 与 admission；
|
||||
- 带 `resourceVersion` 的乐观并发、list/watch 与断线恢复；
|
||||
- informer/cache/workqueue 生态；
|
||||
- authentication、RBAC、namespace、审计与 API discovery；
|
||||
- spec/status、conditions、finalizer 等控制面约定。
|
||||
|
||||
这项选择不把 Ayatori 限定为容器编排系统，也不意味着原生 Kubernetes workload API 是领域
|
||||
模型。kube-apiserver 保存期望、引用和观察状态；Ayatori controller-manager 实现平台领域的
|
||||
调度、生命周期、故障恢复、垃圾回收和后端收敛。Controller 可以运行于专用 management
|
||||
environment，并管理集群外的 VM、LB、数据库、对象存储、DNS、凭据和托管 Kubernetes 控制面。
|
||||
|
||||
Ayatori 可以选择性复用 Kubernetes 内置资源的 API contract，而不采用其上游实现组件。例如，
|
||||
`core/v1 Node` 可以表达计算节点身份、capacity、conditions、labels、taints 和维护状态，由
|
||||
Ayatori Compute Agent 更新并由 Ayatori controller 消费；这不要求部署或模拟 kubelet，也不
|
||||
要求存在 Pod、CRI、kube-scheduler 或 kube-controller-manager。`Lease`、`Namespace`、
|
||||
`Secret`、`ConfigMap`、`Event` 和 RBAC 等资源同样按各自适用的 API 语义独立选择。
|
||||
|
||||
复用内置资源前必须明确其 producer、consumer、ownership、采用的字段和未采用的上游语义。
|
||||
不能因为 Kubernetes 通常将若干组件一起部署，就把这些实现关系重新带入 Ayatori。
|
||||
|
||||
Kubernetes workload 集群与 OpenSandbox、Proxmox 等一样，是通过 adapter 接入的 backend 或
|
||||
executor。它可以是远端集群，也可以完全不存在。除 Flux 和 Ayatori controllers 等管理组件的
|
||||
部署外，领域 API 不得隐含依赖 controller 所在集群的 Pod、Job、Service、NetworkPolicy、
|
||||
namespace 共置或 owner reference 语义；确有需要的能力必须由领域 API 和 adapter 契约显式表达。
|
||||
|
||||
GitOps 是长期期望状态的主要提交入口；API 是当前意图、关系和状态的在线控制面；真实后端
|
||||
仍是运行事实来源。Controller 负责三者之间持续收敛。
|
||||
|
||||
`generic-apiserver` 或 Kubernetes API aggregation 只会让 Ayatori 接管资源的服务端实现，并不会
|
||||
替代上述领域 controller。除非 CRD/kube-apiserver 的存储模型、API 语义或扩展边界形成经过验证的
|
||||
阻碍，Ayatori 不自行承担 watch、RBAC、API 兼容、存储版本迁移和高可用 API Server 的实现与运维。
|
||||
|
||||
## 结果
|
||||
|
||||
- 获得统一声明式 API、watch、RBAC、admission、conditions 和 controller 生态。
|
||||
- Ayatori controller-manager 实际承担类似 kube-controller-manager 的领域控制循环职责，必须把
|
||||
reconcile、状态迁移、恢复与后端契约作为产品核心，而不是把它们误交给 kube-apiserver。
|
||||
- 原生 Kubernetes workload 对象不能成为所有 adapter 的最低公共语义；Kubernetes 只是其中一种
|
||||
执行后端。
|
||||
- 允许由 Ayatori 自己实现合适的内置 API 资源语义；API 类型与上游 controller/runtime 不绑定。
|
||||
- 可以把机器与人工执行统一建模为异步控制循环。
|
||||
- 必须维护 CRD 版本、conversion、认证、备份和控制面升级。
|
||||
- 不在 API 中保存日志、指标、大对象或业务数据，只保存控制所需状态及引用。
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
# ADR-0006：按实际管理缺口扩展资源 API
|
||||
|
||||
- 状态：Accepted
|
||||
- 日期：2026-09-20
|
||||
|
||||
## 背景
|
||||
|
||||
Ayatori 可以在技术上逐步加入 VM、任务、数据库、负载均衡、对象存储、KaaS、FaaS 与应用
|
||||
托管等能力。如果按传统私有云产品目录推进，项目会把后端“能够实现”的能力误当成 homelab
|
||||
实际需要的产品，并承担没有消费者的 API、controller、升级和恢复成本。
|
||||
|
||||
当前真正反复出现的问题，是 Database、LoadBalancer 和 Bucket/Object Storage 缺少符合本环境
|
||||
需求的稳定管理 API。Proxmox VM 也存在明确缺口：远程 API 能力有限，一部分操作只能登录节点
|
||||
使用 CLI 完成，因此单靠 Terraform provider 或 Proxmox API 无法覆盖期望生命周期。
|
||||
|
||||
当前 `Job` controller 是验证 Kubernetes API machinery、状态机、finalizer、回收和 adapter 边界
|
||||
的首个纵向切片。OpenSandbox 和 microVM 可以成为内部执行后端，但这不等于平台需要 Lambda、
|
||||
Cloud Run 或其他 FaaS/PaaS 产品。
|
||||
|
||||
## 决策
|
||||
|
||||
Ayatori 不设置必须完成的云产品清单。新增北向资源必须由现实消费者、重复管理缺口和持续
|
||||
reconcile 的明确收益驱动。
|
||||
|
||||
当前优先方向是：
|
||||
|
||||
1. `Database`；
|
||||
2. `LoadBalancer`；
|
||||
3. `Bucket` / Object Storage；
|
||||
4. `VirtualMachine`，其价值已确认，但实现成本更高。
|
||||
|
||||
`Run`/当前实验性的 `Job` 定位为控制面执行原语和架构验证切片，不自动扩展为面向用户的计算
|
||||
产品。KaaS 是可能有真实需求的候选能力，但不是必达终点。FaaS、Cloud Run 和应用托管默认不做，
|
||||
除非未来以新的需求和 ADR 改变决定。
|
||||
|
||||
VirtualMachine controller 对外提供稳定北向 API；南向允许根据操作选择 Proxmox API、节点上的
|
||||
受限强类型 Agent/CLI 或 `ManualTask`。节点 Agent 必须提供版本化、幂等、可观察和可审计的操作，
|
||||
不能退化为任意远程 shell。
|
||||
|
||||
## 结果
|
||||
|
||||
- 路线图可以根据当前收益调整，不把技术可行性误作产品承诺。
|
||||
- 第一个 Job controller 的实现仍有测试和架构验证价值，但其 API 不约束长期产品形态。
|
||||
- VM 被保留为核心高价值方向，同时承认其南向集成不是单一 provider 能解决的问题。
|
||||
- 每个新增资源都要独立证明生命周期和管理价值；已有 backend 不自动产生新的产品层。
|
||||
@@ -0,0 +1,61 @@
|
||||
# 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 的产品优先级。
|
||||
@@ -0,0 +1,92 @@
|
||||
# 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 保留入口。
|
||||
+23
-25
@@ -9,49 +9,47 @@
|
||||
- 定义 API、conditions、ownership 和 executor 公共约定。
|
||||
- 建立不可变制品与 Dev 到 Prod promotion。
|
||||
|
||||
## 1. Job Service
|
||||
## 1. Controller 纵向验证切片
|
||||
|
||||
- 实现最小 `Job` API。
|
||||
- Kubernetes Pod executor。
|
||||
- 统一日志、退出状态、超时、workspace、cache 与 artifact。
|
||||
- 接入 Gitea Actions 和平台内部 IaC 执行。
|
||||
- 使用当前最小 `Job` API 验证 watch、状态机、finalizer、取消、TTL、external reference 与
|
||||
backend adapter。
|
||||
- Kubernetes executor 不能与 management API client 或同集群 namespace 语义绑定。
|
||||
- 验证完成后，将可复用机制收敛为内部 `Run`/execution 能力；不把这一切片扩展为 FaaS、
|
||||
Cloud Run 或通用 Job Service。
|
||||
|
||||
## 2. OpenSandbox Executor
|
||||
## 2. 首批资源产品
|
||||
|
||||
- 通过 OpenSandbox lifecycle 与 execd API 创建、执行和清理 sandbox。
|
||||
- 支持强隔离任务、未知代码、嵌套容器和 AI agent。
|
||||
- 增加交互式 `Sandbox` API、TTL、endpoint 与 snapshot。
|
||||
- `Database`：PostgreSQL database、role、credential 与回收。
|
||||
- `LoadBalancer`：Envoy 配置/xDS、健康检查、固定 VIP 与 GoBGP 路由宣告。
|
||||
- `Bucket`：SeaweedFS bucket、policy、credential 与删除策略。
|
||||
- 按纵向价值选择先后，不为三者预先建立统一 provider 框架。
|
||||
|
||||
## 3. Human Executor
|
||||
## 3. Human Executor 与延迟自动化
|
||||
|
||||
- `ManualTask`、`TaskReport` 和版本化 Runbook。
|
||||
- Telegram/Email 通知、领取、提醒和升级。
|
||||
- 后端验证与上游 reconcile 恢复。
|
||||
|
||||
## 4. LBaaS
|
||||
## 4. Compute 与节点生命周期
|
||||
|
||||
- Envoy 配置/xDS adapter。
|
||||
- 健康检查与 GoBGP 路由宣告。
|
||||
- 固定 VIP、listener/backend 引用和故障恢复。
|
||||
|
||||
## 5. Compute 与节点生命周期
|
||||
|
||||
- Proxmox VM adapter 与现有资源 adopt。
|
||||
- ComputeNode 加入、drain 和 `SafeToRemove`。
|
||||
- 建立稳定的 `VirtualMachine` 北向 API，并支持现有资源 adopt。
|
||||
- 南向按能力组合 Proxmox API、节点受限 Agent/CLI 与 `ManualTask`，不假设 Proxmox API 完整。
|
||||
- Node 加入、drain 和 `SafeToRemove`；Node API 由 Ayatori Compute Agent 实现，不依赖 kubelet。
|
||||
- StorageClass、StoragePool、Volume 与迁移计划。
|
||||
- 先支持人工磁盘迁移，再通过 Job executor 自动化。
|
||||
- 先支持人工磁盘迁移，再按实际收益自动化。
|
||||
|
||||
## 6. 数据服务
|
||||
## 5. 条件性扩展
|
||||
|
||||
- PostgreSQL database/role/credential。
|
||||
- SeaweedFS bucket/policy/credential。
|
||||
- DNS 与证书资源。
|
||||
- OpenSandbox/microVM 可以作为内部 Run backend，但不由此产生 FaaS 产品承诺。
|
||||
- DNS、证书和 Credential 只有在跨系统协调收益明确时形成独立资源。
|
||||
- KaaS 只有出现托管控制面、租户隔离或频繁集群生命周期的真实需求时才立项。
|
||||
|
||||
## 7. KaaS
|
||||
### KaaS 候选方案
|
||||
|
||||
- 采用成熟 hosted-control-plane 后端。
|
||||
- 组合控制面、worker、LB、DNS、网络和凭据。
|
||||
- 用户集群只暴露 worker node，控制面完全由平台托管。
|
||||
- 本节记录候选实现边界，不构成路线图承诺。
|
||||
|
||||
## 首个业务里程碑
|
||||
|
||||
|
||||
@@ -29,6 +29,20 @@ Ayatori 是具有产品质量的内部平台，而非初期即面向公众的通
|
||||
平台允许对当前环境形成明确意见：Proxmox、OpenSandbox、Envoy、GoBGP、OpenBao、
|
||||
PostgreSQL、SeaweedFS、Samba AD DNS、Cloudflare 和 Flux 都可以是已知实现。
|
||||
|
||||
Ayatori 不以补齐传统私有云或公有云的产品目录为目标。一个资源只有同时满足以下条件，才进入
|
||||
北向 API：
|
||||
|
||||
1. homelab 存在现实消费者和重复需求；
|
||||
2. 现有后端 API 或 IaC 无法提供足够的管理体验；
|
||||
3. 持续 observe/reconcile 明显优于一次性自动化；
|
||||
4. 统一生命周期、状态、组合或权限能产生可验证的收益；
|
||||
5. 收益足以承担长期 API 兼容、controller 和恢复测试成本。
|
||||
|
||||
当前最明确的管理缺口是 Database、LoadBalancer 与 Bucket/Object Storage。VirtualMachine 同样
|
||||
具有明确价值：Proxmox 的 API 不能覆盖所需的全部生命周期，一部分操作必须在节点上通过 CLI
|
||||
完成，因此 Ayatori 可以提供稳定北向 API，并在南向组合 Proxmox API、受限节点 Agent 与人工
|
||||
任务。KaaS 只有在出现托管控制面的实际需求时才进入实现，不是产品路线的必达终点。
|
||||
|
||||
## 非目标
|
||||
|
||||
- 不替代 hypervisor、microVM runtime、数据库、对象存储或网络协议栈。
|
||||
@@ -36,3 +50,5 @@ PostgreSQL、SeaweedFS、Samba AD DNS、Cloudflare 和 Flux 都可以是已知
|
||||
- 不以隐藏全部后端信息或制造虚假多云可移植性为目标。
|
||||
- 不创建理解所有应用需求的中央 Application controller。
|
||||
- 不要求所有人工步骤立即自动化。
|
||||
- 不因为已有 Run、OpenSandbox 或 microVM backend，就构建 FaaS、Cloud Run 或应用托管产品。
|
||||
- 不预先承诺 KaaS；它是需求驱动的候选能力。
|
||||
|
||||
@@ -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