Author SHA1 Message Date
panxiao81 ff21c5bfba docs: 保留 DBaaS 已批准设计合同
Verify / test (pull_request) Successful in 7m34s
Verify / lint (pull_request) Successful in 8m8s
2026-09-20 20:30:39 +00:00
panxiao81 c004aca1cc docs: 统一 Database API 到 Ayatori 域
Verify / test (pull_request) Successful in 6m7s
Verify / lint (pull_request) Successful in 6m32s
2026-09-20 20:20:49 +00:00
panxiao81 85b61a6a23 docs: 允许 Database 模块无兼容负担重构
Verify / test (pull_request) Successful in 6m19s
Verify / lint (pull_request) Successful in 6m43s
2026-09-20 20:11:41 +00:00
panxiao81 4206cfbab7 docs: 记录 Compute 方向与 Database 模块合并
Verify / test (pull_request) Successful in 8m2s
Verify / lint (pull_request) Successful in 9m34s
2026-09-20 19:59:53 +00:00
panxiao81 09a0a5665f docs: 解耦内置 API 与上游实现组件
Verify / test (pull_request) Successful in 8m25s
Verify / lint (pull_request) Successful in 9m0s
2026-09-20 19:42:39 +00:00
panxiao81 d9d99b2f9b docs: 按实际管理缺口限定产品范围 2026-09-20 19:18:12 +00:00
panxiao81 3e21de5942 docs: 明确 API machinery 与领域控制循环边界 2026-09-20 19:02:06 +00:00
panxiao81 f76978ca17 docs: require integration tests for controllers
Verify / test (pull_request) Successful in 11m35s
Verify / lint (pull_request) Successful in 11m58s
2026-09-18 18:33:11 +00:00
panxiao81 06bc54e3cf test: exercise controller with envtest
Verify / test (pull_request) Successful in 8m22s
Verify / lint (pull_request) Successful in 9m37s
2026-09-18 18:30:14 +00:00
panxiao81 86effb72a8 feat: execute jobs on Kubernetes
Verify / test (pull_request) Successful in 7m25s
Verify / lint (pull_request) Successful in 8m9s
2026-09-18 18:20:10 +00:00
9 changed files with 1110 additions and 85 deletions
-76
View File
@@ -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.
+8
View File
@@ -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
View File
@@ -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
+2 -3
View File
@@ -51,10 +51,9 @@
- 用户集群只暴露 worker node,控制面完全由平台托管。
- 本节记录候选实现边界,不构成路线图承诺。
## 后续 Compute 验收场景
## 首个业务里程碑
Database 等首批资源优先落地。Compute 开始实施后,以 Laptop Rebuild Readiness 验证节点
生命周期与恢复能力;该场景不作为首批 Database、LoadBalancer 或 Bucket 的交付前置条件:
完成 Laptop Rebuild Readiness:
1. 临时节点加入。
2. laptop 上的 workload 被重建、迁移或形成可执行人工任务。
+132
View File
@@ -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
}
+71
View File
@@ -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")
}
}
+397
View File
@@ -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 &copy
}
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
}
+245
View File
@@ -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
}