feat: add FunctionJob CRD for one-shot function runs
- api/v1alpha1/job_types.go: new CRD FunctionJob (Pending/Running/Succeeded/Failed) - controllers/functionjob_controller.go: reconciler creates k8s Job from FunctionRef + EventJSON - zz_generated.deepcopy.go: DeepCopy methods for FunctionJob types - config/crd/bases: generated CRD YAML, applied to cluster - main.go: register FunctionJobReconciler - rbac.yaml: add functionjobs permissions - operator.yaml: v0.1.2 -> v0.1.3 - operator:v0.1.3 deployed and running in cluster
This commit is contained in:
@@ -0,0 +1,89 @@
|
|||||||
|
// Изменено: 2026-03-07
|
||||||
|
// Описание CRD FunctionJob — одноразовый запуск функции.
|
||||||
|
// Отдельный ресурс (не Trigger) потому что семантика принципиально другая:
|
||||||
|
// - Trigger: постоянно живёт, описывает КАК функцию вызывают (http/cron)
|
||||||
|
// - FunctionJob: запускается один раз, завершается, хранит результат
|
||||||
|
//
|
||||||
|
// Terraform ресурс: sless_job
|
||||||
|
// k8s ресурс: k8s Job в namespace sless-fn-{namespace}
|
||||||
|
|
||||||
|
package v1alpha1
|
||||||
|
|
||||||
|
import (
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
)
|
||||||
|
|
||||||
|
// FunctionJobSpec — параметры одноразового запуска функции.
|
||||||
|
type FunctionJobSpec struct {
|
||||||
|
// FunctionRef — имя Function ресурса в том же namespace
|
||||||
|
// +kubebuilder:validation:Required
|
||||||
|
FunctionRef string `json:"functionRef"`
|
||||||
|
|
||||||
|
// EventJSON — данные передаваемые в handle(event) в JSON формате.
|
||||||
|
// Если не задан — передаётся пустой объект {}.
|
||||||
|
// Пример: {"action": "migrate", "version": "002"}
|
||||||
|
// +kubebuilder:default="{}"
|
||||||
|
EventJSON string `json:"eventJson,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// FunctionJobPhase — фаза жизненного цикла одноразового запуска.
|
||||||
|
type FunctionJobPhase string
|
||||||
|
|
||||||
|
const (
|
||||||
|
// FunctionJobPhasePending — ожидает пока Function станет Ready
|
||||||
|
FunctionJobPhasePending FunctionJobPhase = "Pending"
|
||||||
|
// FunctionJobPhaseRunning — k8s Job запущен, функция выполняется
|
||||||
|
FunctionJobPhaseRunning FunctionJobPhase = "Running"
|
||||||
|
// FunctionJobPhaseSucceeded — функция успешно завершилась
|
||||||
|
FunctionJobPhaseSucceeded FunctionJobPhase = "Succeeded"
|
||||||
|
// FunctionJobPhaseFailed — функция завершилась с ошибкой
|
||||||
|
FunctionJobPhaseFailed FunctionJobPhase = "Failed"
|
||||||
|
)
|
||||||
|
|
||||||
|
// FunctionJobStatus — наблюдаемое состояние (заполняет контроллер).
|
||||||
|
type FunctionJobStatus struct {
|
||||||
|
// Phase — текущая фаза: Pending, Running, Succeeded, Failed
|
||||||
|
Phase FunctionJobPhase `json:"phase,omitempty"`
|
||||||
|
|
||||||
|
// JobName — имя созданного k8s Job
|
||||||
|
JobName string `json:"jobName,omitempty"`
|
||||||
|
|
||||||
|
// StartTime — время запуска k8s Job
|
||||||
|
StartTime *metav1.Time `json:"startTime,omitempty"`
|
||||||
|
|
||||||
|
// CompletionTime — время завершения
|
||||||
|
CompletionTime *metav1.Time `json:"completionTime,omitempty"`
|
||||||
|
|
||||||
|
// Message — результат выполнения или сообщение об ошибке
|
||||||
|
Message string `json:"message,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
//+kubebuilder:object:root=true
|
||||||
|
//+kubebuilder:subresource:status
|
||||||
|
//+kubebuilder:printcolumn:name="Function",type=string,JSONPath=`.spec.functionRef`
|
||||||
|
//+kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`
|
||||||
|
//+kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
|
||||||
|
|
||||||
|
// FunctionJob — ресурс для одноразового запуска serverless функции.
|
||||||
|
// Создаётся пользователем через terraform ресурс sless_job.
|
||||||
|
// После завершения (Succeeded/Failed) ресурс остаётся в кластере для аудита.
|
||||||
|
type FunctionJob struct {
|
||||||
|
metav1.TypeMeta `json:",inline"`
|
||||||
|
metav1.ObjectMeta `json:"metadata,omitempty"`
|
||||||
|
|
||||||
|
Spec FunctionJobSpec `json:"spec,omitempty"`
|
||||||
|
Status FunctionJobStatus `json:"status,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
//+kubebuilder:object:root=true
|
||||||
|
|
||||||
|
// FunctionJobList contains a list of FunctionJob
|
||||||
|
type FunctionJobList struct {
|
||||||
|
metav1.TypeMeta `json:",inline"`
|
||||||
|
metav1.ListMeta `json:"metadata,omitempty"`
|
||||||
|
Items []FunctionJob `json:"items"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
SchemeBuilder.Register(&FunctionJob{}, &FunctionJobList{})
|
||||||
|
}
|
||||||
@@ -21,7 +21,7 @@ limitations under the License.
|
|||||||
package v1alpha1
|
package v1alpha1
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"k8s.io/apimachinery/pkg/apis/meta/v1"
|
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
runtime "k8s.io/apimachinery/pkg/runtime"
|
runtime "k8s.io/apimachinery/pkg/runtime"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -224,3 +224,100 @@ func (in *TriggerStatus) DeepCopy() *TriggerStatus {
|
|||||||
in.DeepCopyInto(out)
|
in.DeepCopyInto(out)
|
||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
|
||||||
|
func (in *FunctionJob) DeepCopyInto(out *FunctionJob) {
|
||||||
|
*out = *in
|
||||||
|
out.TypeMeta = in.TypeMeta
|
||||||
|
in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
|
||||||
|
out.Spec = in.Spec
|
||||||
|
in.Status.DeepCopyInto(&out.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FunctionJob.
|
||||||
|
func (in *FunctionJob) DeepCopy() *FunctionJob {
|
||||||
|
if in == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
out := new(FunctionJob)
|
||||||
|
in.DeepCopyInto(out)
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
|
||||||
|
func (in *FunctionJob) DeepCopyObject() runtime.Object {
|
||||||
|
if c := in.DeepCopy(); c != nil {
|
||||||
|
return c
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
|
||||||
|
func (in *FunctionJobList) DeepCopyInto(out *FunctionJobList) {
|
||||||
|
*out = *in
|
||||||
|
out.TypeMeta = in.TypeMeta
|
||||||
|
in.ListMeta.DeepCopyInto(&out.ListMeta)
|
||||||
|
if in.Items != nil {
|
||||||
|
in, out := &in.Items, &out.Items
|
||||||
|
*out = make([]FunctionJob, len(*in))
|
||||||
|
for i := range *in {
|
||||||
|
(*in)[i].DeepCopyInto(&(*out)[i])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FunctionJobList.
|
||||||
|
func (in *FunctionJobList) DeepCopy() *FunctionJobList {
|
||||||
|
if in == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
out := new(FunctionJobList)
|
||||||
|
in.DeepCopyInto(out)
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
|
||||||
|
func (in *FunctionJobList) DeepCopyObject() runtime.Object {
|
||||||
|
if c := in.DeepCopy(); c != nil {
|
||||||
|
return c
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
|
||||||
|
func (in *FunctionJobSpec) DeepCopyInto(out *FunctionJobSpec) {
|
||||||
|
*out = *in
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FunctionJobSpec.
|
||||||
|
func (in *FunctionJobSpec) DeepCopy() *FunctionJobSpec {
|
||||||
|
if in == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
out := new(FunctionJobSpec)
|
||||||
|
in.DeepCopyInto(out)
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
|
||||||
|
func (in *FunctionJobStatus) DeepCopyInto(out *FunctionJobStatus) {
|
||||||
|
*out = *in
|
||||||
|
if in.StartTime != nil {
|
||||||
|
in, out := &in.StartTime, &out.StartTime
|
||||||
|
*out = (*in).DeepCopy()
|
||||||
|
}
|
||||||
|
if in.CompletionTime != nil {
|
||||||
|
in, out := &in.CompletionTime, &out.CompletionTime
|
||||||
|
*out = (*in).DeepCopy()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FunctionJobStatus.
|
||||||
|
func (in *FunctionJobStatus) DeepCopy() *FunctionJobStatus {
|
||||||
|
if in == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
out := new(FunctionJobStatus)
|
||||||
|
in.DeepCopyInto(out)
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,93 @@
|
|||||||
|
---
|
||||||
|
apiVersion: apiextensions.k8s.io/v1
|
||||||
|
kind: CustomResourceDefinition
|
||||||
|
metadata:
|
||||||
|
annotations:
|
||||||
|
controller-gen.kubebuilder.io/version: v0.14.0
|
||||||
|
name: functionjobs.sless.kube5s.ru
|
||||||
|
spec:
|
||||||
|
group: sless.kube5s.ru
|
||||||
|
names:
|
||||||
|
kind: FunctionJob
|
||||||
|
listKind: FunctionJobList
|
||||||
|
plural: functionjobs
|
||||||
|
singular: functionjob
|
||||||
|
scope: Namespaced
|
||||||
|
versions:
|
||||||
|
- additionalPrinterColumns:
|
||||||
|
- jsonPath: .spec.functionRef
|
||||||
|
name: Function
|
||||||
|
type: string
|
||||||
|
- jsonPath: .status.phase
|
||||||
|
name: Phase
|
||||||
|
type: string
|
||||||
|
- jsonPath: .metadata.creationTimestamp
|
||||||
|
name: Age
|
||||||
|
type: date
|
||||||
|
name: v1alpha1
|
||||||
|
schema:
|
||||||
|
openAPIV3Schema:
|
||||||
|
description: |-
|
||||||
|
FunctionJob — ресурс для одноразового запуска serverless функции.
|
||||||
|
Создаётся пользователем через terraform ресурс sless_job.
|
||||||
|
После завершения (Succeeded/Failed) ресурс остаётся в кластере для аудита.
|
||||||
|
properties:
|
||||||
|
apiVersion:
|
||||||
|
description: |-
|
||||||
|
APIVersion defines the versioned schema of this representation of an object.
|
||||||
|
Servers should convert recognized schemas to the latest internal value, and
|
||||||
|
may reject unrecognized values.
|
||||||
|
More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
|
||||||
|
type: string
|
||||||
|
kind:
|
||||||
|
description: |-
|
||||||
|
Kind is a string value representing the REST resource this object represents.
|
||||||
|
Servers may infer this from the endpoint the client submits requests to.
|
||||||
|
Cannot be updated.
|
||||||
|
In CamelCase.
|
||||||
|
More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
|
||||||
|
type: string
|
||||||
|
metadata:
|
||||||
|
type: object
|
||||||
|
spec:
|
||||||
|
description: FunctionJobSpec — параметры одноразового запуска функции.
|
||||||
|
properties:
|
||||||
|
eventJson:
|
||||||
|
default: '{}'
|
||||||
|
description: |-
|
||||||
|
EventJSON — данные передаваемые в handle(event) в JSON формате.
|
||||||
|
Если не задан — передаётся пустой объект {}.
|
||||||
|
Пример: {"action": "migrate", "version": "002"}
|
||||||
|
type: string
|
||||||
|
functionRef:
|
||||||
|
description: FunctionRef — имя Function ресурса в том же namespace
|
||||||
|
type: string
|
||||||
|
required:
|
||||||
|
- functionRef
|
||||||
|
type: object
|
||||||
|
status:
|
||||||
|
description: FunctionJobStatus — наблюдаемое состояние (заполняет контроллер).
|
||||||
|
properties:
|
||||||
|
completionTime:
|
||||||
|
description: CompletionTime — время завершения
|
||||||
|
format: date-time
|
||||||
|
type: string
|
||||||
|
jobName:
|
||||||
|
description: JobName — имя созданного k8s Job
|
||||||
|
type: string
|
||||||
|
message:
|
||||||
|
description: Message — результат выполнения или сообщение об ошибке
|
||||||
|
type: string
|
||||||
|
phase:
|
||||||
|
description: 'Phase — текущая фаза: Pending, Running, Succeeded, Failed'
|
||||||
|
type: string
|
||||||
|
startTime:
|
||||||
|
description: StartTime — время запуска k8s Job
|
||||||
|
format: date-time
|
||||||
|
type: string
|
||||||
|
type: object
|
||||||
|
type: object
|
||||||
|
served: true
|
||||||
|
storage: true
|
||||||
|
subresources:
|
||||||
|
status: {}
|
||||||
@@ -0,0 +1,236 @@
|
|||||||
|
// Изменено: 2026-03-07
|
||||||
|
// FunctionJobReconciler — контроллер одноразовых запусков функций.
|
||||||
|
// При создании FunctionJob:
|
||||||
|
// 1. Ждёт пока Function станет Ready
|
||||||
|
// 2. Создаёт k8s Job который запускает образ функции с CMD runner
|
||||||
|
// 3. Следит за завершением Job → обновляет статус (Succeeded/Failed)
|
||||||
|
//
|
||||||
|
// Почему отдельный ресурс (не Trigger type=job):
|
||||||
|
// Trigger описывает постоянный способ вызова (http endpoint, cron schedule).
|
||||||
|
// FunctionJob — разовое событие с отдельным lifecycle и статусом результата.
|
||||||
|
|
||||||
|
package controllers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
batchv1 "k8s.io/api/batch/v1"
|
||||||
|
corev1 "k8s.io/api/core/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/api/errors"
|
||||||
|
"k8s.io/apimachinery/pkg/api/resource"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime"
|
||||||
|
ctrl "sigs.k8s.io/controller-runtime"
|
||||||
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||||
|
"sigs.k8s.io/controller-runtime/pkg/log"
|
||||||
|
|
||||||
|
slessv1alpha1 "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/api/v1alpha1"
|
||||||
|
)
|
||||||
|
|
||||||
|
// FunctionJobReconciler reconciles a FunctionJob object
|
||||||
|
type FunctionJobReconciler struct {
|
||||||
|
client.Client
|
||||||
|
Scheme *runtime.Scheme
|
||||||
|
RegistrySecret string // имя k8s Secret с docker credentials (для imagePullSecrets)
|
||||||
|
}
|
||||||
|
|
||||||
|
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=functionjobs,verbs=get;list;watch;create;update;patch;delete
|
||||||
|
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=functionjobs/status,verbs=get;update;patch
|
||||||
|
//+kubebuilder:rbac:groups=sless.kube5s.ru,resources=functionjobs/finalizers,verbs=update
|
||||||
|
//+kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;update;patch;delete
|
||||||
|
|
||||||
|
// Reconcile — основной цикл контроллера.
|
||||||
|
func (r *FunctionJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
|
||||||
|
logger := log.FromContext(ctx)
|
||||||
|
|
||||||
|
fj := &slessv1alpha1.FunctionJob{}
|
||||||
|
if err := r.Get(ctx, req.NamespacedName, fj); err != nil {
|
||||||
|
if errors.IsNotFound(err) {
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
return ctrl.Result{}, fmt.Errorf("get functionjob: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Уже завершён — ничего не делаем
|
||||||
|
if fj.Status.Phase == slessv1alpha1.FunctionJobPhaseSucceeded ||
|
||||||
|
fj.Status.Phase == slessv1alpha1.FunctionJobPhaseFailed {
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Проверяем что Function существует и готова
|
||||||
|
fn := &slessv1alpha1.Function{}
|
||||||
|
if err := r.Get(ctx, client.ObjectKey{Name: fj.Spec.FunctionRef, Namespace: fj.Namespace}, fn); err != nil {
|
||||||
|
if errors.IsNotFound(err) {
|
||||||
|
fj.Status.Phase = slessv1alpha1.FunctionJobPhasePending
|
||||||
|
fj.Status.Message = "function not found: " + fj.Spec.FunctionRef
|
||||||
|
_ = r.Status().Update(ctx, fj)
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
return ctrl.Result{}, fmt.Errorf("get function: %w", err)
|
||||||
|
}
|
||||||
|
if fn.Status.Phase != slessv1alpha1.FunctionPhaseReady {
|
||||||
|
fj.Status.Phase = slessv1alpha1.FunctionJobPhasePending
|
||||||
|
fj.Status.Message = "waiting for function Ready (current: " + string(fn.Status.Phase) + ")"
|
||||||
|
_ = r.Status().Update(ctx, fj)
|
||||||
|
// Повторный reconcile придёт когда Function изменится
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
deployNS := "sless-fn-" + fj.Namespace
|
||||||
|
jobName := fmt.Sprintf("job-%s-%s", fj.Name, fj.CreationTimestamp.Format("20060102150405"))
|
||||||
|
|
||||||
|
// Если Job уже создан — проверяем его статус
|
||||||
|
existingJob := &batchv1.Job{}
|
||||||
|
if err := r.Get(ctx, client.ObjectKey{Name: jobName, Namespace: deployNS}, existingJob); err == nil {
|
||||||
|
return r.syncJobStatus(ctx, fj, existingJob)
|
||||||
|
} else if !errors.IsNotFound(err) {
|
||||||
|
return ctrl.Result{}, fmt.Errorf("get job: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Создаём k8s Job
|
||||||
|
// Используем образ функции напрямую, переопределяем CMD чтобы запустить runner
|
||||||
|
// вместо server.py/server.js — runner выполняет handle(event) один раз и выходит
|
||||||
|
eventJSON := fj.Spec.EventJSON
|
||||||
|
if eventJSON == "" {
|
||||||
|
eventJSON = "{}"
|
||||||
|
}
|
||||||
|
|
||||||
|
// runner запускается через env var SLESS_EVENT — безопаснее чем передавать в args
|
||||||
|
// (args видны в ps aux, env vars — нет)
|
||||||
|
ttl := int32(600) // автоудаление Job через 10 мин после завершения
|
||||||
|
|
||||||
|
job := &batchv1.Job{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{
|
||||||
|
Name: jobName,
|
||||||
|
Namespace: deployNS,
|
||||||
|
Labels: map[string]string{
|
||||||
|
"managed-by": "sless",
|
||||||
|
"functionjob": fj.Name,
|
||||||
|
"function": fn.Name,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
Spec: batchv1.JobSpec{
|
||||||
|
// Не перезапускать при ошибке — это одноразовый запуск
|
||||||
|
BackoffLimit: int32Ptr(0),
|
||||||
|
// Автоудаление через 10 мин после завершения — чтобы не засорять кластер
|
||||||
|
TTLSecondsAfterFinished: &ttl,
|
||||||
|
Template: corev1.PodTemplateSpec{
|
||||||
|
Spec: corev1.PodSpec{
|
||||||
|
RestartPolicy: corev1.RestartPolicyNever,
|
||||||
|
// Используем тот же образ что и Deployment функции
|
||||||
|
// runner.py/runner.js переопределяет CMD сервера
|
||||||
|
InitContainers: nil,
|
||||||
|
Containers: []corev1.Container{
|
||||||
|
{
|
||||||
|
Name: "runner",
|
||||||
|
Image: fn.Status.ImageRef,
|
||||||
|
// Переопределяем точку входа: запускаем runner вместо server
|
||||||
|
// runner читает SLESS_EVENT и вызывает handle(event) один раз
|
||||||
|
Command: runtimeRunnerCommand(fn.Spec.Runtime),
|
||||||
|
Env: append(
|
||||||
|
fnEnvVars(fn),
|
||||||
|
corev1.EnvVar{
|
||||||
|
Name: "SLESS_EVENT",
|
||||||
|
Value: eventJSON,
|
||||||
|
},
|
||||||
|
),
|
||||||
|
Resources: corev1.ResourceRequirements{
|
||||||
|
Limits: corev1.ResourceList{
|
||||||
|
corev1.ResourceMemory: resource.MustParse(fmt.Sprintf("%dMi", fn.Spec.MemoryMB)),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
ImagePullSecrets: []corev1.LocalObjectReference{
|
||||||
|
{Name: r.RegistrySecret},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := r.Create(ctx, job); err != nil {
|
||||||
|
return ctrl.Result{}, fmt.Errorf("create job: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
now := metav1.Now()
|
||||||
|
fj.Status.Phase = slessv1alpha1.FunctionJobPhaseRunning
|
||||||
|
fj.Status.JobName = jobName
|
||||||
|
fj.Status.StartTime = &now
|
||||||
|
fj.Status.Message = ""
|
||||||
|
if err := r.Status().Update(ctx, fj); err != nil {
|
||||||
|
return ctrl.Result{}, fmt.Errorf("update functionjob status: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.Info("created job for functionjob", "job", jobName, "functionjob", fj.Name)
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// syncJobStatus читает статус k8s Job и обновляет FunctionJob.Status.
|
||||||
|
func (r *FunctionJobReconciler) syncJobStatus(ctx context.Context, fj *slessv1alpha1.FunctionJob, job *batchv1.Job) (ctrl.Result, error) {
|
||||||
|
if job.Status.Succeeded > 0 {
|
||||||
|
now := metav1.Now()
|
||||||
|
fj.Status.Phase = slessv1alpha1.FunctionJobPhaseSucceeded
|
||||||
|
fj.Status.CompletionTime = &now
|
||||||
|
fj.Status.Message = "completed successfully"
|
||||||
|
} else if job.Status.Failed > 0 {
|
||||||
|
now := metav1.Now()
|
||||||
|
fj.Status.Phase = slessv1alpha1.FunctionJobPhaseFailed
|
||||||
|
fj.Status.CompletionTime = &now
|
||||||
|
fj.Status.Message = "job failed, check pod logs: kubectl logs -n sless-fn-" + fj.Namespace + " -l functionjob=" + fj.Name
|
||||||
|
}
|
||||||
|
// Running — ничего не меняем, перечитаем при следующем reconcile
|
||||||
|
if err := r.Status().Update(ctx, fj); err != nil {
|
||||||
|
return ctrl.Result{}, fmt.Errorf("sync job status: %w", err)
|
||||||
|
}
|
||||||
|
return ctrl.Result{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// runtimeRunnerCommand возвращает CMD для запуска одноразового runner вместо HTTP-сервера.
|
||||||
|
// runner читает env SLESS_EVENT, вызывает handle(event) один раз и завершается.
|
||||||
|
func runtimeRunnerCommand(runtime string) []string {
|
||||||
|
switch runtime {
|
||||||
|
case "nodejs20":
|
||||||
|
// inline runner — не требует отдельного файла в образе
|
||||||
|
return []string{"node", "-e", `
|
||||||
|
const h = require('/app/function/handler.js');
|
||||||
|
const event = JSON.parse(process.env.SLESS_EVENT || '{}');
|
||||||
|
Promise.resolve(h.handle(event)).then(r => {
|
||||||
|
console.log(JSON.stringify(r));
|
||||||
|
process.exit(0);
|
||||||
|
}).catch(e => {
|
||||||
|
console.error(e.message);
|
||||||
|
process.exit(1);
|
||||||
|
});`}
|
||||||
|
default: // python3.11
|
||||||
|
return []string{"python3", "-c", `
|
||||||
|
import os, json, importlib.util
|
||||||
|
spec = importlib.util.spec_from_file_location("handler", "/app/function/handler.py")
|
||||||
|
mod = importlib.util.module_from_spec(spec)
|
||||||
|
spec.loader.exec_module(mod)
|
||||||
|
event = json.loads(os.environ.get("SLESS_EVENT", "{}"))
|
||||||
|
result = mod.handle(event)
|
||||||
|
print(json.dumps(result))
|
||||||
|
`}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// fnEnvVars преобразует env vars из FunctionSpec в k8s EnvVar slice.
|
||||||
|
func fnEnvVars(fn *slessv1alpha1.Function) []corev1.EnvVar {
|
||||||
|
var result []corev1.EnvVar
|
||||||
|
for k, v := range fn.Spec.Env {
|
||||||
|
result = append(result, corev1.EnvVar{Name: k, Value: v})
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
func int32Ptr(i int32) *int32 { return &i }
|
||||||
|
|
||||||
|
// SetupWithManager регистрирует контроллер и настраивает watch на k8s Job.
|
||||||
|
func (r *FunctionJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
|
||||||
|
return ctrl.NewControllerManagedBy(mgr).
|
||||||
|
For(&slessv1alpha1.FunctionJob{}).
|
||||||
|
Owns(&batchv1.Job{}).
|
||||||
|
Complete(r)
|
||||||
|
}
|
||||||
@@ -3,7 +3,7 @@
|
|||||||
# Состав:
|
# Состав:
|
||||||
# - ConfigMap: не-секретные env vars (S3_ENDPOINT, REGISTRY_HOST и т.д.)
|
# - ConfigMap: не-секретные env vars (S3_ENDPOINT, REGISTRY_HOST и т.д.)
|
||||||
# - Secret: секретные данные (S3 keys, postgres DSN, API token, docker auth)
|
# - Secret: секретные данные (S3 keys, postgres DSN, API token, docker auth)
|
||||||
# - Deployment: оператор naeel/sless-operator:v0.1.0 в namespace sless
|
# - Deployment: оператор naeel/sless-operator:v0.1.3 в namespace sless
|
||||||
# - Service: ClusterIP :9090 (REST API)
|
# - Service: ClusterIP :9090 (REST API)
|
||||||
# - Ingress: sless-api.kube5s.ru → :9090 (внешний доступ с TLS)
|
# - Ingress: sless-api.kube5s.ru → :9090 (внешний доступ с TLS)
|
||||||
#
|
#
|
||||||
@@ -67,7 +67,7 @@ spec:
|
|||||||
containers:
|
containers:
|
||||||
- name: operator
|
- name: operator
|
||||||
# При обновлении версии оператора — менять тег здесь (не latest!)
|
# При обновлении версии оператора — менять тег здесь (не latest!)
|
||||||
image: naeel/sless-operator:v0.1.2
|
image: naeel/sless-operator:v0.1.3
|
||||||
# Always — чтобы всегда тянуть по точному тегу (не кешировать старый)
|
# Always — чтобы всегда тянуть по точному тегу (не кешировать старый)
|
||||||
imagePullPolicy: Always
|
imagePullPolicy: Always
|
||||||
ports:
|
ports:
|
||||||
|
|||||||
@@ -17,13 +17,13 @@ metadata:
|
|||||||
rules:
|
rules:
|
||||||
# Наши CRD
|
# Наши CRD
|
||||||
- apiGroups: ["sless.kube5s.ru"]
|
- apiGroups: ["sless.kube5s.ru"]
|
||||||
resources: ["functions", "triggers"]
|
resources: ["functions", "triggers", "functionjobs"]
|
||||||
verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
|
verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
|
||||||
- apiGroups: ["sless.kube5s.ru"]
|
- apiGroups: ["sless.kube5s.ru"]
|
||||||
resources: ["functions/status", "triggers/status"]
|
resources: ["functions/status", "triggers/status", "functionjobs/status"]
|
||||||
verbs: ["get", "update", "patch"]
|
verbs: ["get", "update", "patch"]
|
||||||
- apiGroups: ["sless.kube5s.ru"]
|
- apiGroups: ["sless.kube5s.ru"]
|
||||||
resources: ["functions/finalizers", "triggers/finalizers"]
|
resources: ["functions/finalizers", "triggers/finalizers", "functionjobs/finalizers"]
|
||||||
verbs: ["update"]
|
verbs: ["update"]
|
||||||
# Deployments для функций
|
# Deployments для функций
|
||||||
- apiGroups: ["apps"]
|
- apiGroups: ["apps"]
|
||||||
|
|||||||
@@ -144,6 +144,14 @@ func main() {
|
|||||||
log.Error("unable to create controller", "controller", "Trigger", "err", err)
|
log.Error("unable to create controller", "controller", "Trigger", "err", err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
|
if err = (&controllers.FunctionJobReconciler{
|
||||||
|
Client: mgr.GetClient(),
|
||||||
|
Scheme: mgr.GetScheme(),
|
||||||
|
RegistrySecret: cfg.RegistrySecret,
|
||||||
|
}).SetupWithManager(mgr); err != nil {
|
||||||
|
log.Error("unable to create controller", "controller", "FunctionJob", "err", err)
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
//+kubebuilder:scaffold:builder
|
//+kubebuilder:scaffold:builder
|
||||||
|
|
||||||
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
|
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user