Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
86953cd91a
|
||
|
|
ddd717209e
|
||
|
|
2756803ba4
|
||
|
|
76752c8443
|
||
|
|
36138f835a
|
||
|
|
de8b7f9b04
|
||
|
|
e6b9980b3e
|
||
|
|
641db531a8
|
||
|
|
7b841d7dba
|
@@ -0,0 +1,76 @@
|
||||
---
|
||||
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.
|
||||
@@ -20,7 +20,6 @@ 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
|
||||
)
|
||||
|
||||
@@ -166,13 +165,6 @@ 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 {
|
||||
|
||||
+6
-53
@@ -1,58 +1,11 @@
|
||||
---
|
||||
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:
|
||||
- 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
|
||||
- apiGroups: [""]
|
||||
resources: ["pods"]
|
||||
verbs: ["get", "list", "watch"]
|
||||
|
||||
+3
-2
@@ -51,9 +51,10 @@
|
||||
- 用户集群只暴露 worker node,控制面完全由平台托管。
|
||||
- 本节记录候选实现边界,不构成路线图承诺。
|
||||
|
||||
## 首个业务里程碑
|
||||
## 后续 Compute 验收场景
|
||||
|
||||
完成 Laptop Rebuild Readiness:
|
||||
Database 等首批资源优先落地。Compute 开始实施后,以 Laptop Rebuild Readiness 验证节点
|
||||
生命周期与恢复能力;该场景不作为首批 Database、LoadBalancer 或 Bucket 的交付前置条件:
|
||||
|
||||
1. 临时节点加入。
|
||||
2. laptop 上的 workload 被重建、迁移或形成可执行人工任务。
|
||||
|
||||
@@ -1,132 +0,0 @@
|
||||
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
|
||||
}
|
||||
@@ -1,71 +0,0 @@
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -1,397 +0,0 @@
|
||||
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)
|
||||
}
|
||||
@@ -1,202 +0,0 @@
|
||||
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
|
||||
}
|
||||
@@ -1,245 +0,0 @@
|
||||
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