diff --git a/api/v1alpha1/job_types.go b/api/v1alpha1/job_types.go new file mode 100644 index 0000000..bc8122b --- /dev/null +++ b/api/v1alpha1/job_types.go @@ -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{}) +} diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 1623f3c..c51a146 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -21,7 +21,7 @@ limitations under the License. package v1alpha1 import ( - "k8s.io/apimachinery/pkg/apis/meta/v1" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" runtime "k8s.io/apimachinery/pkg/runtime" ) @@ -224,3 +224,100 @@ func (in *TriggerStatus) DeepCopy() *TriggerStatus { in.DeepCopyInto(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 +} diff --git a/config/crd/bases/sless.kube5s.ru_functionjobs.yaml b/config/crd/bases/sless.kube5s.ru_functionjobs.yaml new file mode 100644 index 0000000..318034b --- /dev/null +++ b/config/crd/bases/sless.kube5s.ru_functionjobs.yaml @@ -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: {} diff --git a/controllers/functionjob_controller.go b/controllers/functionjob_controller.go new file mode 100644 index 0000000..6b94c6e --- /dev/null +++ b/controllers/functionjob_controller.go @@ -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) +} diff --git a/deployments/k8s/operator.yaml b/deployments/k8s/operator.yaml index 03aa0f9..fa1fddb 100644 --- a/deployments/k8s/operator.yaml +++ b/deployments/k8s/operator.yaml @@ -3,7 +3,7 @@ # Состав: # - ConfigMap: не-секретные env vars (S3_ENDPOINT, REGISTRY_HOST и т.д.) # - 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) # - Ingress: sless-api.kube5s.ru → :9090 (внешний доступ с TLS) # @@ -67,7 +67,7 @@ spec: containers: - name: operator # При обновлении версии оператора — менять тег здесь (не latest!) - image: naeel/sless-operator:v0.1.2 + image: naeel/sless-operator:v0.1.3 # Always — чтобы всегда тянуть по точному тегу (не кешировать старый) imagePullPolicy: Always ports: diff --git a/deployments/k8s/rbac.yaml b/deployments/k8s/rbac.yaml index eb220e1..c08d59e 100644 --- a/deployments/k8s/rbac.yaml +++ b/deployments/k8s/rbac.yaml @@ -17,13 +17,13 @@ metadata: rules: # Наши CRD - apiGroups: ["sless.kube5s.ru"] - resources: ["functions", "triggers"] + resources: ["functions", "triggers", "functionjobs"] verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] - apiGroups: ["sless.kube5s.ru"] - resources: ["functions/status", "triggers/status"] + resources: ["functions/status", "triggers/status", "functionjobs/status"] verbs: ["get", "update", "patch"] - apiGroups: ["sless.kube5s.ru"] - resources: ["functions/finalizers", "triggers/finalizers"] + resources: ["functions/finalizers", "triggers/finalizers", "functionjobs/finalizers"] verbs: ["update"] # Deployments для функций - apiGroups: ["apps"] diff --git a/main.go b/main.go index 047ac35..d4e0583 100644 --- a/main.go +++ b/main.go @@ -144,6 +144,14 @@ func main() { log.Error("unable to create controller", "controller", "Trigger", "err", err) 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 if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {