Nail/sless
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(v0.1.58): ImageExists cache hit, timing analysis, in-cluster registry plan

Browse Source
This commit is contained in:
Naeel
2026-03-23 06:22:25 +03:00
parent 7023e0e6fc
commit 9edd43edc5
20 changed files with 596 additions and 515 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/handler/functions.go
+73 -11
View File
@@ -1,5 +1,6 @@
// Изменено: 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}
@@ -9,6 +10,7 @@ package handler
import (
"encoding/json"
"net/http"
"time"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -128,22 +130,59 @@ func (h *Handler) CreateFunction(w http.ResponseWriter, r *http.Request) {
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) {
// IsAlreadyExists может прийти из кеша controller-runtime (split-brain):
// объект удалён из etcd, но кеш informer ещё южив. Делаем uncached Get:
// если реально NotFound — кеш устарел, пересоздаём.
// если существует и фаза Failed — тоже пересоздаём (build провалился, терраформ не добавил в state).
// если существует и фаза Ready/Building — возвращаем 409 (функция реально есть).
// Получаем актуальное состояние объекта из etcd (не из кеша informer).
existing := &slessv1alpha1.Function{}
getErr := h.K8s.Get(r.Context(), client.ObjectKey{Name: req.Name, Namespace: ns}, existing)
shouldRecreate := errors.IsNotFound(getErr) ||
(getErr == nil && existing.Status.Phase == slessv1alpha1.FunctionPhaseFailed)
if shouldRecreate {
if getErr == nil {
_ = h.K8s.Delete(r.Context(), 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) — продолжаем ждать.
}
// Сбрасываем ResourceVersion — при split-brain etcd считает объект новым
// Таймаут: объект не исчез за 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()))
@@ -152,6 +191,29 @@ func (h *Handler) CreateFunction(w http.ResponseWriter, r *http.Request) {
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
}
internal/api/handler/services.go
+89 -9
View File
@@ -1,5 +1,6 @@
// Создано: 2026-03-20 (function-service-split)
// Изменено: 2026-03-21 (fix: DeleteService возвращает 404 вместо 204 при отсутствующем объекте)
// Изменено: 2026-03-22 (fix: CreateService 409 при пересоздании сервиса через terraform -replace)
// services.go — CRUD handlers для Service CRD (sless_service).
// sless_service = long-running Deployment + URL. Каждый вызов проксируется к поду.
// Namespace берётся из URL: /v1/namespaces/{namespace}/services/{name}
@@ -134,18 +135,69 @@ func (h *Handler) CreateService(w http.ResponseWriter, r *http.Request) {
S3Key: req.S3Key,
},
}
// Пытаемся создать Service CRD в k8s.
// h.K8s.Create обращается к etcd через controller-runtime client.
if err := h.K8s.Create(r.Context(), svc); err != nil {
// IsAlreadyExists = etcd вернул 409 CONFLICT.
// Это нормально при конкурентных запросах или при "replace" через terraform.
// Сценарии:
//
// [A] terraform apply -replace:
// 1. terraform DELETE /services/{name} → API вызывает h.K8s.Delete(svc)
// 2. k8s НЕ удаляет объект немедленно: ставит DeletionTimestamp и ждёт
// пока service_controller.go снимет finalizer sless.kube5s.ru/service-finalizer
// (контроллер сначала сносит Deployment/Service/Ingress)
// 3. terraform сразу POST /services/{name} → h.K8s.Create → IsAlreadyExists
// (объект ещё есть в etcd, просто помечен на удаление)
// → СТАРЫЙ КОД: видел phase=Ready, shouldRecreate=false → возвращал 409 ← БАГ
// → НОВЫЙ КОД: видит DeletionTimestamp → ждёт исчезновения → создаёт
//
// [B] Кеш controller-runtime (split-brain):
// Объект удалён из etcd, но informer-кеш ещё не обновился.
// h.K8s.Create падает с IsAlreadyExists из кеша, хотя в etcd объекта нет.
// → Get возвращает NotFound → пересоздаём.
//
// [C] Сервис в фазе Failed:
// Предыдущий build провалился, terraform не добавил в state.
// Новый apply пытается создать снова → Failed объект удаляем и пересоздаём.
if errors.IsAlreadyExists(err) {
// Повторяем логику function_handler: обрабатываем split-brain кеша.
// Если объект реально есть и не в фазе Failed — возвращаем 409.
// Получаем актуальное состояние объекта напрямую из etcd (не из кеша).
// Нужно чтобы точно определить сценарий A/B/C.
existing := &slessv1alpha1.Service{}
getErr := h.K8s.Get(r.Context(), client.ObjectKey{Name: req.Name, Namespace: ns}, existing)
shouldRecreate := errors.IsNotFound(getErr) ||
(getErr == nil && existing.Status.Phase == slessv1alpha1.ServicePhaseFailed)
if shouldRecreate {
if getErr == nil {
_ = h.K8s.Delete(r.Context(), existing)
// --- Сценарий A: объект помечен на удаление ---
// DeletionTimestamp ≠ zero означает что k8s принял DELETE и ждёт finalizer.
// Finalizer снимает service_controller.go асинхронно (обычно 1-5 сек).
// Мы не можем создать объект пока старый существует — ждём его исчезновения.
if getErr == nil && !existing.DeletionTimestamp.IsZero() {
deleted := false
// Проверяем каждую секунду до 30 итераций (= 30 секунд максимум).
// Обычно finalizer снимается за 1-3 секунды, 30 — запас на перегруженный кластер.
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)
// IsNotFound = объект полностью исчез из etcd — можно создавать.
if errors.IsNotFound(checkErr) {
deleted = true
break
}
// Любая другая ошибка Get — продолжаем ждать (может быть временный сбой API).
}
// 30 секунд прошло, объект всё ещё не удалён.
// Вероятно контроллер завис или finalizer не снимается.
// Возвращаем 409 с понятным сообщением — клиент (terraform) должен retry.
if !deleted {
writeJSON(w, http.StatusConflict, errResp("service is being deleted, try again later"))
return
}
// Объект исчез. Сбрасываем ResourceVersion чтобы k8s воспринял как новый объект.
// ResourceVersion заполняется при первом Get — при Create должен быть пустым.
svc.ResourceVersion = ""
if createErr := h.K8s.Create(r.Context(), svc); createErr != nil {
writeJSON(w, http.StatusInternalServerError, errResp(createErr.Error()))
@@ -154,11 +206,39 @@ func (h *Handler) CreateService(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusCreated, svcToResponse(svc))
return
}
// --- Сценарии B и C: split-brain кеша или объект в фазе Failed ---
// shouldRecreate=true если:
// - getErr = NotFound (кеш врёт — объекта в etcd нет)
// - объект есть, но в фазе Failed (предыдущий build провалился)
shouldRecreate := errors.IsNotFound(getErr) ||
(getErr == nil && existing.Status.Phase == slessv1alpha1.ServicePhaseFailed)
if shouldRecreate {
// Если объект реально существует (Failed) — удаляем его перед пересозданием.
// Ошибку удаления игнорируем: даже если Delete упал,
// следующий Create либо пройдёт (объект исчез) либо снова упадёт с понятной ошибкой.
if getErr == nil {
_ = h.K8s.Delete(r.Context(), existing)
}
// Сбрасываем ResourceVersion — при split-brain etcd считает объект новым.
svc.ResourceVersion = ""
if createErr := h.K8s.Create(r.Context(), svc); createErr != nil {
writeJSON(w, http.StatusInternalServerError, errResp(createErr.Error()))
return
}
writeJSON(w, http.StatusCreated, svcToResponse(svc))
return
}
// Объект существует, DeletionTimestamp=zero, фаза не Failed.
// Это легитимный конфликт — сервис реально работает, клиент пытается создать дубликат.
writeJSON(w, http.StatusConflict, errResp("service already exists"))
return
}
// k8s возвращает StatusInvalid (422) при нарушении enum-валидации CRD (например, неизвестный runtime)
// — маппим это в 400, а не в 500
// k8s возвращает StatusInvalid (422) при нарушении enum-валидации CRD.
// Например: runtime="java8" которого нет в openapi enum сервисного CRD.
// Маппим в 400 (Bad Request) а не в 500 — это ошибка клиента, не сервера.
if errors.IsInvalid(err) {
writeJSON(w, http.StatusBadRequest, errResp("invalid service spec: "+err.Error()))
return
internal/builder/builder.go
+107 -19
View File
@@ -1,16 +1,22 @@
// Изменено: 2026-03-11
// Изменено: 2026-03-22
// Builder — запускает kaniko Job в k8s для сборки Docker образа из кода функции.
// Почему kaniko, а не docker-in-docker (DinD):
// kaniko не требует privileged контейнер, что безопаснее в managed кластере.
// kaniko читает контекст сборки из S3 напрямую.
// Workflow: FunctionController вызывает Build → EnsureProject (Harbor) → Job запускается → образ пушится в registry.
// ImageExists: перед сборкой проверяем наличие образа по тегу в registry.
// Если образ с таким тегом (= hash кода) уже существует — сборка пропускается.
// Это даёт идемпотентность destroy+apply: если код не менялся, deploy мгновенный.
package builder
import (
"context"
"crypto/sha256"
"encoding/json"
"fmt"
"net/http"
"strings"
"time"
batchv1 "k8s.io/api/batch/v1"
@@ -33,12 +39,12 @@ type Builder struct {
registryHost string // куда пушим образ (DockerHub: "naeel"; Harbor: "host")
registryProject string // проект/org внутри registry (Harbor: project; DockerHub: пусто)
registrySecret string // имя k8s Secret с docker-кредами для kaniko
s3Endpoint string // откуда kaniko берёт код
s3AccessKey string
s3SecretKey string
s3Bucket string
namespace string // namespace где запускаем build Job'ы
harborClient Projecter // nil если Harbor не используется
s3Endpoint string // откуда kaniko берёт код
s3AccessKey string
s3SecretKey string
s3Bucket string
namespace string // namespace где запускаем build Job'ы
harborClient Projecter // nil если Harbor не используется
}
// Config — параметры для создания Builder'а.
@@ -47,12 +53,12 @@ type Config struct {
RegistryHost string
RegistryProject string // пусто = DockerHub-режим (2 уровня); задан = project-режим (3 уровня)
RegistrySecret string // имя k8s Secret с .dockerconfigjson для пуша образов
S3Endpoint string
S3AccessKey string
S3SecretKey string
S3Bucket string
Namespace string
HarborClient Projecter // nil — Harbor не используется, EnsureProject пропускается
S3Endpoint string
S3AccessKey string
S3SecretKey string
S3Bucket string
Namespace string
HarborClient Projecter // nil — Harbor не используется, EnsureProject пропускается
}
// New создаёт новый Builder.
@@ -63,12 +69,12 @@ func New(c client.Client, cfg Config) *Builder {
registryHost: cfg.RegistryHost,
registryProject: cfg.RegistryProject,
registrySecret: cfg.RegistrySecret,
s3Endpoint: cfg.S3Endpoint,
s3AccessKey: cfg.S3AccessKey,
s3SecretKey: cfg.S3SecretKey,
s3Bucket: cfg.S3Bucket,
namespace: cfg.Namespace,
harborClient: cfg.HarborClient,
s3Endpoint: cfg.S3Endpoint,
s3AccessKey: cfg.S3AccessKey,
s3SecretKey: cfg.S3SecretKey,
s3Bucket: cfg.S3Bucket,
namespace: cfg.Namespace,
harborClient: cfg.HarborClient,
}
}
@@ -103,6 +109,88 @@ func (b *Builder) ImageRef(namespace, funcName, s3Key string) string {
return fmt.Sprintf("%s/%s-%s:%s", b.registryHost, nsPrefix, funcName, tag)
}
// ImageExists проверяет наличие образа в Docker Registry v2 по тегу (без pull).
// Использует анонимный bearer-token для публичных репо (DockerHub).
// Возвращает true если образ с таким тегом уже запушен — сборка не нужна.
//
// Почему анонимный токен, а не credentials:
//
// DockerHub выдаёт pull-token без авторизации для публичных репо через
// GET /token?service=registry.docker.io&scope=repository:{repo}:pull
// Это стандартный Docker Registry v2 auth flow (RFC 7235).
func (b *Builder) ImageExists(ctx context.Context, imageRef string) bool {
// imageRef вида: "naeel/slessffd1-pg-search:47cab27ada70"
// или "host/project/func:tag" — разбираем по последнему ":"
colonIdx := strings.LastIndex(imageRef, ":")
if colonIdx < 0 {
return false
}
repoFull := imageRef[:colonIdx]
tag := imageRef[colonIdx+1:]
// Определяем registry host и repo path.
// DockerHub: "naeel/sless-ff-pg" → registry = index.docker.io, repo = "naeel/sless-ff-pg"
// Приватный: "harbor.host/proj/func" → registry = "harbor.host", repo = "proj/func"
registryHost := "index.docker.io"
repo := repoFull
if parts := strings.SplitN(repoFull, "/", 3); len(parts) == 3 {
// host/project/name — кастомный registry
registryHost = parts[0]
repo = parts[1] + "/" + parts[2]
}
// Получаем анонимный/публичный bearer-token для pull доступа к репо.
// DockerHub: https://auth.docker.io/token?service=registry.docker.io&scope=repository:{repo}:pull
// Для приватных registry этот шаг вернёт 401 → ImageExists вернёт false → пойдём строить.
tokenURL := fmt.Sprintf("https://auth.docker.io/token?service=registry.docker.io&scope=repository:%s:pull", repo)
if registryHost != "index.docker.io" {
// Для не-DockerHub registry: пробуем без токена (Harbor с allow anon push)
// Если 401 — false, пусть builder разберётся.
tokenURL = ""
}
httpClient := &http.Client{Timeout: 5 * time.Second}
var bearerToken string
if tokenURL != "" {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, tokenURL, nil)
if err != nil {
return false
}
resp, err := httpClient.Do(req)
if err != nil || resp.StatusCode != http.StatusOK {
return false
}
defer resp.Body.Close()
var tkResp struct {
Token string `json:"token"`
}
if err := json.NewDecoder(resp.Body).Decode(&tkResp); err != nil {
return false
}
bearerToken = tkResp.Token
}
// HEAD /v2/{repo}/manifests/{tag} — проверяем наличие тега без скачивания слоёв.
manifestURL := fmt.Sprintf("https://%s/v2/%s/manifests/%s", registryHost, repo, tag)
req, err := http.NewRequestWithContext(ctx, http.MethodHead, manifestURL, nil)
if err != nil {
return false
}
// Docker Registry v2 требует Accept header для манифестов.
req.Header.Set("Accept", "application/vnd.docker.distribution.manifest.v2+json")
if bearerToken != "" {
req.Header.Set("Authorization", "Bearer "+bearerToken)
}
resp, err := httpClient.Do(req)
if err != nil {
return false
}
defer resp.Body.Close()
return resp.StatusCode == http.StatusOK
}
// Build запускает kaniko Job для сборки образа функции.
// Возвращает имя Job'а чтобы контроллер мог следить за его статусом.
func (b *Builder) Build(ctx context.Context, namespace, funcName, s3Key string) (string, error) {