311 lines
13 KiB
Go
311 lines
13 KiB
Go
// Изменено: 2026-03-18 (добавлены created_at, last_built_at в functionResponse и fnToResponse)
|
||
// Изменено: 2026-03-21 (fix: DeleteFunction возвращает 404 вместо 204 при отсутствующем объекте)
|
||
// Изменено: 2026-03-22 (fix: CreateFunction 409 при пересоздании функции — не учитывался DeletionTimestamp)
|
||
// functions.go — CRUD handlers для Function CRD.
|
||
// Принимает JSON, создаёт/обновляет/удаляет k8s ресурсы Function.
|
||
// Namespace берётся из URL: /v1/namespaces/{namespace}/functions/{name}
|
||
|
||
package handler
|
||
|
||
import (
|
||
"encoding/json"
|
||
"net/http"
|
||
"time"
|
||
|
||
"k8s.io/apimachinery/pkg/api/errors"
|
||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||
|
||
slessv1alpha1 "gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/api/v1alpha1"
|
||
)
|
||
|
||
// functionRequest — тело запроса для создания/обновления функции.
|
||
type functionRequest struct {
|
||
Name string `json:"name"`
|
||
Runtime string `json:"runtime"`
|
||
Entrypoint string `json:"entrypoint"`
|
||
MemoryMB int32 `json:"memory_mb"`
|
||
TimeoutSec int32 `json:"timeout_sec"`
|
||
Env map[string]string `json:"env_vars"`
|
||
S3Bucket string `json:"s3_bucket"`
|
||
S3Key string `json:"s3_key"`
|
||
}
|
||
|
||
// functionResponse — ответ при чтении функции.
|
||
type functionResponse struct {
|
||
Name string `json:"name"`
|
||
Namespace string `json:"namespace"`
|
||
Runtime string `json:"runtime"`
|
||
Entrypoint string `json:"entrypoint"`
|
||
MemoryMB int32 `json:"memory_mb"`
|
||
TimeoutSec int32 `json:"timeout_sec"`
|
||
Env map[string]string `json:"env_vars"`
|
||
S3Bucket string `json:"s3_bucket"`
|
||
S3Key string `json:"s3_key"`
|
||
Phase slessv1alpha1.FunctionPhase `json:"phase"`
|
||
ImageRef string `json:"image_ref"`
|
||
Message string `json:"message,omitempty"`
|
||
// CreatedAt — время создания CRD объекта (metadata.creationTimestamp).
|
||
// Пустое значение = "0001-01-01T00:00:00Z" сериализуется в "", опускаем через omitempty.
|
||
CreatedAt string `json:"created_at,omitempty"`
|
||
// LastBuiltAt — время последней успешной сборки образа (status.lastBuiltAt).
|
||
// nil если сборки ещё не было.
|
||
LastBuiltAt string `json:"last_built_at,omitempty"`
|
||
}
|
||
|
||
// fnToResponse конвертирует CRD в ответ API.
|
||
func fnToResponse(fn *slessv1alpha1.Function) functionResponse {
|
||
resp := functionResponse{
|
||
Name: fn.Name,
|
||
Namespace: fn.Namespace,
|
||
Runtime: fn.Spec.Runtime,
|
||
Entrypoint: fn.Spec.Entrypoint,
|
||
MemoryMB: fn.Spec.MemoryMB,
|
||
TimeoutSec: fn.Spec.TimeoutSec,
|
||
Env: fn.Spec.Env,
|
||
S3Bucket: fn.Spec.S3Bucket,
|
||
S3Key: fn.Spec.S3Key,
|
||
Phase: fn.Status.Phase,
|
||
ImageRef: fn.Status.ImageRef,
|
||
Message: fn.Status.Message,
|
||
}
|
||
// creationTimestamp — всегда заполнен k8s, но zero value опускаем.
|
||
if !fn.CreationTimestamp.IsZero() {
|
||
resp.CreatedAt = fn.CreationTimestamp.UTC().Format("2006-01-02 15:04:05 UTC")
|
||
}
|
||
if fn.Status.LastBuiltAt != nil && !fn.Status.LastBuiltAt.IsZero() {
|
||
resp.LastBuiltAt = fn.Status.LastBuiltAt.UTC().Format("2006-01-02 15:04:05 UTC")
|
||
}
|
||
return resp
|
||
}
|
||
|
||
// ListFunctions — GET /v1/namespaces/{namespace}/functions
|
||
func (h *Handler) ListFunctions(w http.ResponseWriter, r *http.Request) {
|
||
ns := namespace(r)
|
||
list := &slessv1alpha1.FunctionList{}
|
||
if err := h.K8s.List(r.Context(), list, client.InNamespace(ns)); err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||
return
|
||
}
|
||
result := make([]functionResponse, 0, len(list.Items))
|
||
for i := range list.Items {
|
||
result = append(result, fnToResponse(&list.Items[i]))
|
||
}
|
||
writeJSON(w, http.StatusOK, result)
|
||
}
|
||
|
||
// CreateFunction — POST /v1/namespaces/{namespace}/functions
|
||
func (h *Handler) CreateFunction(w http.ResponseWriter, r *http.Request) {
|
||
ns := namespace(r)
|
||
var req functionRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, errResp("invalid JSON: "+err.Error()))
|
||
return
|
||
}
|
||
if req.Name == "" || req.Runtime == "" {
|
||
writeJSON(w, http.StatusBadRequest, errResp("name and runtime are required"))
|
||
return
|
||
}
|
||
if req.Entrypoint == "" {
|
||
writeJSON(w, http.StatusBadRequest, errResp("entrypoint is required"))
|
||
return
|
||
}
|
||
if req.MemoryMB <= 0 || req.MemoryMB > 4096 {
|
||
writeJSON(w, http.StatusBadRequest, errResp("memory_mb must be between 1 and 4096"))
|
||
return
|
||
}
|
||
|
||
fn := &slessv1alpha1.Function{
|
||
ObjectMeta: metav1.ObjectMeta{
|
||
Name: req.Name,
|
||
Namespace: ns,
|
||
},
|
||
Spec: slessv1alpha1.FunctionSpec{
|
||
Runtime: req.Runtime,
|
||
Entrypoint: req.Entrypoint,
|
||
MemoryMB: req.MemoryMB,
|
||
TimeoutSec: req.TimeoutSec,
|
||
Env: req.Env,
|
||
S3Bucket: req.S3Bucket,
|
||
S3Key: req.S3Key,
|
||
},
|
||
}
|
||
// Пытаемся создать Function CRD в k8s (запись в etcd через controller-runtime).
|
||
if err := h.K8s.Create(r.Context(), fn); err != nil {
|
||
|
||
// IsAlreadyExists = etcd вернул 409 CONFLICT.
|
||
// Возникает в двух основных сценариях:
|
||
//
|
||
// [A] terraform apply -replace (= delete + create за один apply):
|
||
// 1. terraform DELETE /functions/{name} → API вызывает h.K8s.Delete(fn)
|
||
// 2. k8s ставит DeletionTimestamp и ждёт снятия finalizer sless.kube5s.ru/finalizer
|
||
// function_controller.go убивает kaniko Job и снимает finalizer асинхронно
|
||
// 3. terraform сразу POST /functions/{name} → IsAlreadyExists ← БАГ до этого фикса
|
||
// → НОВЫЙ КОД: обнаруживает DeletionTimestamp → ждёт исчезновения → создаёт
|
||
//
|
||
// [B] Split-brain кеша controller-runtime:
|
||
// Объект удалён из etcd, но informer-кеш ещё не обновился.
|
||
// Create падает с IsAlreadyExists из кеша, Get возвращает NotFound → пересоздаём.
|
||
//
|
||
// [C] Функция в фазе Failed:
|
||
// Предыдущий build провалился. Новый apply пытается создать снова.
|
||
// Удаляем Failed объект и пересоздаём.
|
||
if errors.IsAlreadyExists(err) {
|
||
|
||
// Получаем актуальное состояние объекта из etcd (не из кеша informer).
|
||
existing := &slessv1alpha1.Function{}
|
||
getErr := h.K8s.Get(r.Context(), client.ObjectKey{Name: req.Name, Namespace: ns}, existing)
|
||
|
||
// --- Сценарий A: объект ожидает удаления ---
|
||
// DeletionTimestamp ≠ zero = k8s принял DELETE, finalizer ещё не снят.
|
||
// function_controller.go снимает finalizer после cleanup Job — обычно 1-5 сек.
|
||
if getErr == nil && !existing.DeletionTimestamp.IsZero() {
|
||
|
||
deleted := false
|
||
// Polling каждую секунду, максимум 30 раз (= 30 секунд).
|
||
// 30 сек — запас на медленный кластер; в норме 1-3 итерации.
|
||
for i := 0; i < 30; i++ {
|
||
time.Sleep(1 * time.Second)
|
||
checkErr := h.K8s.Get(r.Context(), client.ObjectKey{Name: req.Name, Namespace: ns}, existing)
|
||
// NotFound = finalizer снят, объект исчез из etcd — можно создавать.
|
||
if errors.IsNotFound(checkErr) {
|
||
deleted = true
|
||
break
|
||
}
|
||
// Любая другая ошибка (timeout, сбой API) — продолжаем ждать.
|
||
}
|
||
|
||
// Таймаут: объект не исчез за 30 секунд.
|
||
// Клиент (terraform) получит 409 и должен сделать retry позже.
|
||
if !deleted {
|
||
writeJSON(w, http.StatusConflict, errResp("function is being deleted, try again later"))
|
||
return
|
||
}
|
||
|
||
// Объект исчез — сбрасываем ResourceVersion и создаём как новый.
|
||
fn.ResourceVersion = ""
|
||
if createErr := h.K8s.Create(r.Context(), fn); createErr != nil {
|
||
writeJSON(w, http.StatusInternalServerError, errResp(createErr.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusCreated, fnToResponse(fn))
|
||
return
|
||
}
|
||
|
||
// --- Сценарии B и C: split-brain или Failed ---
|
||
// NotFound при Get = кеш врёт (B); Failed фаза = сломанный build (C).
|
||
shouldRecreate := errors.IsNotFound(getErr) ||
|
||
(getErr == nil && existing.Status.Phase == slessv1alpha1.FunctionPhaseFailed)
|
||
if shouldRecreate {
|
||
// Если объект реально есть (Failed) — сначала удаляем.
|
||
// Ошибку Delete игнорируем: Create покажет ошибку сам если что-то пошло не так.
|
||
if getErr == nil {
|
||
_ = h.K8s.Delete(r.Context(), existing)
|
||
}
|
||
// Сбрасываем ResourceVersion — при split-brain etcd считает объект новым.
|
||
fn.ResourceVersion = ""
|
||
if createErr := h.K8s.Create(r.Context(), fn); createErr != nil {
|
||
writeJSON(w, http.StatusInternalServerError, errResp(createErr.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusCreated, fnToResponse(fn))
|
||
return
|
||
}
|
||
|
||
// Объект живой (Ready/Building), DeletionTimestamp=zero.
|
||
// Легитимный конфликт — клиент пытается создать дубликат.
|
||
writeJSON(w, http.StatusConflict, errResp("function already exists"))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusCreated, fnToResponse(fn))
|
||
}
|
||
|
||
// GetFunction — GET /v1/namespaces/{namespace}/functions/{name}
|
||
func (h *Handler) GetFunction(w http.ResponseWriter, r *http.Request) {
|
||
ns := namespace(r)
|
||
name := pathVar(r, "name")
|
||
fn := &slessv1alpha1.Function{}
|
||
if err := h.K8s.Get(r.Context(), client.ObjectKey{Name: name, Namespace: ns}, fn); err != nil {
|
||
if errors.IsNotFound(err) {
|
||
writeJSON(w, http.StatusNotFound, errResp("function not found"))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, fnToResponse(fn))
|
||
}
|
||
|
||
// UpdateFunction — PUT /v1/namespaces/{namespace}/functions/{name}
|
||
func (h *Handler) UpdateFunction(w http.ResponseWriter, r *http.Request) {
|
||
ns := namespace(r)
|
||
name := pathVar(r, "name")
|
||
var req functionRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, errResp("invalid JSON: "+err.Error()))
|
||
return
|
||
}
|
||
|
||
fn := &slessv1alpha1.Function{}
|
||
if err := h.K8s.Get(r.Context(), client.ObjectKey{Name: name, Namespace: ns}, fn); err != nil {
|
||
if errors.IsNotFound(err) {
|
||
writeJSON(w, http.StatusNotFound, errResp("function not found"))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||
return
|
||
}
|
||
|
||
// Валидация обязательных полей — PUT семантика, нельзя обнулять критичные поля
|
||
if req.Runtime == "" || req.Entrypoint == "" {
|
||
writeJSON(w, http.StatusBadRequest, errResp("runtime and entrypoint are required"))
|
||
return
|
||
}
|
||
if req.MemoryMB <= 0 || req.MemoryMB > 4096 {
|
||
writeJSON(w, http.StatusBadRequest, errResp("memory_mb must be between 1 and 4096"))
|
||
return
|
||
}
|
||
|
||
// Обновляем только изменяемые поля spec
|
||
fn.Spec.Runtime = req.Runtime
|
||
fn.Spec.Entrypoint = req.Entrypoint
|
||
fn.Spec.MemoryMB = req.MemoryMB
|
||
fn.Spec.TimeoutSec = req.TimeoutSec
|
||
fn.Spec.Env = req.Env
|
||
if req.S3Bucket != "" {
|
||
fn.Spec.S3Bucket = req.S3Bucket
|
||
}
|
||
if req.S3Key != "" {
|
||
fn.Spec.S3Key = req.S3Key
|
||
}
|
||
|
||
if err := h.K8s.Update(r.Context(), fn); err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, fnToResponse(fn))
|
||
}
|
||
|
||
// DeleteFunction — DELETE /v1/namespaces/{namespace}/functions/{name}
|
||
func (h *Handler) DeleteFunction(w http.ResponseWriter, r *http.Request) {
|
||
ns := namespace(r)
|
||
name := pathVar(r, "name")
|
||
fn := &slessv1alpha1.Function{}
|
||
if err := h.K8s.Get(r.Context(), client.ObjectKey{Name: name, Namespace: ns}, fn); err != nil {
|
||
if errors.IsNotFound(err) {
|
||
writeJSON(w, http.StatusNotFound, errResp("function not found"))
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||
return
|
||
}
|
||
if err := h.K8s.Delete(r.Context(), fn); err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||
return
|
||
}
|
||
w.WriteHeader(http.StatusNoContent)
|
||
}
|