feat(job): merge sless_function into sless_job — self-contained build+run

- FunctionJobSpec: убран FunctionRef, добавлены Runtime/Entrypoint/Env/S3Key/MemoryMB/TimeoutSec - FunctionJobStatus: новый ImageRef, новая фаза Building - FunctionJobReconciler: Building фаза (kaniko), убрана зависимость от Function CRD Builder+OperatorNamespace как поля struct; аннотация sless.kube5s.ru/build-job guard - main.go: Builder+OperatorNamespace переданы в FunctionJobReconciler - jobs.go handler: jobRequest/jobResponse без FunctionRef; новый UploadJobCode handler - router.go: /jobs/{name}/upload маршрут - client.go: JobRequest/JobResponse обновлены; UploadJobCode; uploadCodeToURL общий хелпер - job_resource.go: полная переработка — источник/среда встроены в JobModel, ModifyPlan, Create с upload, wait_timeout_sec=900 по умолчанию (kaniko + выполнение) - examples/POSTGRES/functions.tf: раскомментирован, sless_function удалён, sless_job самодостаточен (inline source_dir/runtime/entrypoint/env_vars)
2026-03-20 21:27:35 +03:00
parent 1b3c8376c0
commit 8ca8faedd1
12 changed files with 569 additions and 185 deletions
@@ -1,8 +1,8 @@
-// Изменено: 2026-03-17 20:00 (bugfix: job-name label удалён в k8s 1.27+, split-brain cached client)
+// Изменено: 2026-03-20 (merge sless_function+sless_job: FunctionJobReconciler самодостаточен)
 // FunctionJobReconciler — контроллер одноразовых запусков функций.
-// При создании FunctionJob:
-//   1. Ждёт пока Function станет Ready
-//   2. Создаёт k8s Job который запускает образ функции с CMD runner
+// При создании FunctionJob с RunID>0:
+//   1. Запускает kaniko сборку образа (фаза Building) — больше не зависит от Function CRD
+//   2. После сборки создаёт k8s Job который запускает образ функции с CMD runner
 //   3. Следит за завершением Job → обновляет статус (Succeeded/Failed)
 //
 // Почему отдельный ресурс (не Trigger type=job):
@@ -31,14 +31,17 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log"

 	slessv1alpha1 "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/api/v1alpha1"
+	"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/builder"
 )

 // FunctionJobReconciler reconciles a FunctionJob object
 type FunctionJobReconciler struct {
 	client.Client
-	Scheme         *runtime.Scheme
-	RegistrySecret string               // имя k8s Secret с docker credentials (для imagePullSecrets)
-	KubeClient     kubernetes.Interface // typed client для чтения логов подов (logs API недоступен через controller-runtime client)
+	Scheme            *runtime.Scheme
+	RegistrySecret    string               // имя k8s Secret с docker credentials (для imagePullSecrets)
+	KubeClient        kubernetes.Interface // typed client для чтения логов подов (logs API недоступен через controller-runtime client)
+	Builder           *builder.Builder     // kaniko builder — собирает образ функции
+	OperatorNamespace string               // namespace оператора — где запускаются kaniko job'ы (обычно "sless")
 }

 //+kubebuilder:rbac:groups=sless.kube5s.ru,resources=functionjobs,verbs=get;list;watch;create;update;patch;delete
@@ -48,8 +51,6 @@ type FunctionJobReconciler struct {

 // 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) {
@@ -76,29 +77,28 @@ func (r *FunctionJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		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)
-		// RequeueAfter: опрашиваем каждые 15с пока Function не станет Ready.
-		return ctrl.Result{RequeueAfter: 15 * time.Second}, nil
+	// Фаза Building — ждём завершения kaniko Job
+	if fj.Status.Phase == slessv1alpha1.FunctionJobPhaseBuilding {
+		return r.checkJobBuild(ctx, fj)
 	}

+	// Нет ImageRef — нужно собрать образ сначала.
+	// Если аннотация build-job уже есть но phase не Building (рестарт контроллера),
+	// восстанавливаем фазу Building.
+	if fj.Status.ImageRef == "" {
+		if fj.Annotations["sless.kube5s.ru/build-job"] != "" {
+			fj.Status.Phase = slessv1alpha1.FunctionJobPhaseBuilding
+			fj.Status.Message = "resuming build: " + fj.Annotations["sless.kube5s.ru/build-job"]
+			_ = r.Status().Update(ctx, fj)
+			return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
+		}
+		return r.startJobBuild(ctx, fj)
+	}
+
+	// Образ собран — создаём/синхронизируем k8s Job
 	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)
@@ -106,14 +106,105 @@ func (r *FunctionJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, fmt.Errorf("get job: %w", err)
 	}

-	// Создаём k8s Job
-	// Используем образ функции напрямую, переопределяем CMD чтобы запустить runner
-	// вместо server.py/server.js — runner выполняет handle(event) один раз и выходит
+	return r.createRunJob(ctx, fj, deployNS, jobName)
+}
+
+// startJobBuild запускает kaniko сборку образа и переводит FunctionJob в фазу Building.
+func (r *FunctionJobReconciler) startJobBuild(ctx context.Context, fj *slessv1alpha1.FunctionJob) (ctrl.Result, error) {
+	logger := log.FromContext(ctx)
+	buildJobName, err := r.Builder.Build(ctx, r.OperatorNamespace, fj.Name, fj.Spec.S3Key)
+	if err != nil {
+		fj.Status.Phase = slessv1alpha1.FunctionJobPhaseFailed
+		fj.Status.Message = "failed to start build: " + err.Error()
+		_ = r.Status().Update(ctx, fj)
+		return ctrl.Result{}, nil
+	}
+
+	// Аннотация build-job guard против повторного запуска при параллельных reconcile.
+	// Как только аннотация выставлена, следующий reconcile войдёт в checkJobBuild.
+	if fj.Annotations == nil {
+		fj.Annotations = map[string]string{}
+	}
+	fj.Annotations["sless.kube5s.ru/build-job"] = buildJobName
+	if err := r.Update(ctx, fj); err != nil {
+		return ctrl.Result{}, fmt.Errorf("update build annotation: %w", err)
+	}
+
+	fj.Status.Phase = slessv1alpha1.FunctionJobPhaseBuilding
+	fj.Status.Message = "building image: " + buildJobName
+	if err := r.Status().Update(ctx, fj); err != nil {
+		return ctrl.Result{}, fmt.Errorf("update status to building: %w", err)
+	}
+
+	logger.Info("started kaniko build for functionjob", "build-job", buildJobName, "functionjob", fj.Name)
+	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
+}
+
+// checkJobBuild опрашивает статус kaniko Job.
+// При успехе: сохраняет ImageRef в status, очищает Build Job, requeue → createRunJob.
+// При ошибке: переводит FunctionJob в Failed с логами kaniko.
+func (r *FunctionJobReconciler) checkJobBuild(ctx context.Context, fj *slessv1alpha1.FunctionJob) (ctrl.Result, error) {
+	logger := log.FromContext(ctx)
+	buildJobName := fj.Annotations["sless.kube5s.ru/build-job"]
+	if buildJobName == "" {
+		fj.Status.Phase = slessv1alpha1.FunctionJobPhasePending
+		_ = r.Status().Update(ctx, fj)
+		return ctrl.Result{Requeue: true}, nil
+	}
+
+	status, err := r.Builder.JobStatus(ctx, buildJobName)
+	if err != nil {
+		return ctrl.Result{}, fmt.Errorf("check build job: %w", err)
+	}
+
+	switch status {
+	case "running":
+		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
+
+	case "succeeded":
+		// Builder вычисляет imageRef детерминировано по namespace+name+s3Key
+		imageRef := r.Builder.ImageRef(r.OperatorNamespace, fj.Name, fj.Spec.S3Key)
+		fj.Status.ImageRef = imageRef
+		// Сбрасываем Phase чтобы следующий reconcile пошёл в createRunJob
+		fj.Status.Phase = slessv1alpha1.FunctionJobPhasePending
+		fj.Status.Message = ""
+		if err := r.Status().Update(ctx, fj); err != nil {
+			return ctrl.Result{}, fmt.Errorf("update imageref in status: %w", err)
+		}
+		_ = r.Builder.Cleanup(ctx, buildJobName)
+		logger.Info("build succeeded, queuing run job", "image", imageRef, "functionjob", fj.Name)
+		return ctrl.Result{Requeue: true}, nil
+
+	case "failed":
+		// Ищем поды kaniko по лейблу job-name в namespace оператора
+		logs := getJobPodOutput(ctx, r.KubeClient, r.OperatorNamespace, "job-name="+buildJobName)
+		msg := "build job failed"
+		if logs != "" {
+			msg = "build job failed:\n" + logs
+		}
+		fj.Status.Phase = slessv1alpha1.FunctionJobPhaseFailed
+		fj.Status.Message = msg
+		_ = r.Status().Update(ctx, fj)
+		return ctrl.Result{}, nil
+	}
+
+	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
+}
+
+// createRunJob создаёт k8s Job для выполнения функции.
+// Использует собранный образ из fj.Status.ImageRef.
+func (r *FunctionJobReconciler) createRunJob(ctx context.Context, fj *slessv1alpha1.FunctionJob, deployNS, jobName string) (ctrl.Result, error) {
+	logger := log.FromContext(ctx)
 	eventJSON := fj.Spec.EventJSON
 	if eventJSON == "" {
 		eventJSON = "{}"
 	}

+	memMB := fj.Spec.MemoryMB
+	if memMB <= 0 {
+		memMB = 128
+	}
+
 	// runner запускается через env var SLESS_EVENT — безопаснее чем передавать в args
 	// (args видны в ps aux, env vars — нет)
 	ttl := int32(600) // автоудаление Job через 10 мин после завершения
@@ -125,7 +216,6 @@ func (r *FunctionJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			Labels: map[string]string{
 				"managed-by":  "sless",
 				"functionjob": fj.Name,
-				"function":    fn.Name,
 			},
 		},
 		Spec: batchv1.JobSpec{
@@ -138,31 +228,25 @@ func (r *FunctionJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 					Labels: map[string]string{
 						"managed-by":  "sless",
 						"functionjob": fj.Name,
-						"function":    fn.Name,
 					},
 				},
 				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),
+							Name:    "runner",
+							Image:   fj.Status.ImageRef,
+							Command: runtimeRunnerCommand(fj.Spec.Runtime),
 							Env: append(
-								append(fnEnvVars(fn), corev1.EnvVar{
+								append(fjEnvVars(fj), corev1.EnvVar{
 									Name:  "SLESS_EVENT",
 									Value: eventJSON,
 								}),
-								goJobModeEnv(fn.Spec.Runtime)...,
+								goJobModeEnv(fj.Spec.Runtime)...,
 							),
 							Resources: corev1.ResourceRequirements{
 								Limits: corev1.ResourceList{
-									corev1.ResourceMemory: resource.MustParse(fmt.Sprintf("%dMi", fn.Spec.MemoryMB)),
+									corev1.ResourceMemory: resource.MustParse(fmt.Sprintf("%dMi", memMB)),
 								},
 							},
 						},
@@ -188,10 +272,11 @@ func (r *FunctionJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, fmt.Errorf("update functionjob status: %w", err)
 	}

-	logger.Info("created job for functionjob", "job", jobName, "functionjob", fj.Name)
+	logger.Info("created run job for functionjob", "job", jobName, "functionjob", fj.Name)
 	return ctrl.Result{}, nil
 }

+
 // syncJobStatus читает статус k8s Job и обновляет FunctionJob.Status.
 // Если Job ещё выполняется — запрашивает повторный reconcile через 5 сек
 // (Owns не работает кросс-неймспейсно, поэтому используем polling).
@@ -282,13 +367,13 @@ print(json.dumps(result))
 	}
 }

-// fnEnvVars преобразует env vars из FunctionSpec в k8s EnvVar slice.
-// Включает SLESS_ENTRYPOINT чтобы runner.py/runner.js знал какую функцию вызывать.
-func fnEnvVars(fn *slessv1alpha1.Function) []corev1.EnvVar {
+// fjEnvVars формирует k8s EnvVar из полей FunctionJobSpec.
+// SLESS_ENTRYPOINT сообщает runner'у какую функцию вызывать.
+func fjEnvVars(fj *slessv1alpha1.FunctionJob) []corev1.EnvVar {
 	result := []corev1.EnvVar{
-		{Name: "SLESS_ENTRYPOINT", Value: fn.Spec.Entrypoint},
+		{Name: "SLESS_ENTRYPOINT", Value: fj.Spec.Entrypoint},
 	}
-	for k, v := range fn.Spec.Env {
+	for k, v := range fj.Spec.Env {
 		result = append(result, corev1.EnvVar{Name: k, Value: v})
 	}
 	return result