From f405596d427bc7b5197e5d0db3899fce2fe92a8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Sat, 7 Mar 2026 17:43:26 +0400 Subject: [PATCH] feat: sless_job terraform resource + jobs REST API Operator (v0.1.4): - internal/api/handler/jobs.go: CreateJob/GetJob/DeleteJob handlers - internal/api/router.go: POST/GET/DELETE /v1/namespaces/{ns}/jobs/{name} Terraform provider (v0.1.2): - client.go: JobRequest/JobResponse + CreateJob/GetJob/DeleteJob/WaitJobDone - resources/job_resource.go: sless_job resource (Create blocks until Succeeded/Failed) - provider.go: register NewJobResource Deploy: - operator:v0.1.4 deployed and running in cluster - provider v0.1.2 published to terra.k8c.ru --- deployments/k8s/operator.yaml | 4 +- internal/api/handler/jobs.go | 149 +++++++++++++ internal/api/router.go | 5 + terraform/provider/internal/client/client.go | 99 ++++++++ .../provider/internal/provider/provider.go | 1 + .../internal/resources/job_resource.go | 211 ++++++++++++++++++ 6 files changed, 467 insertions(+), 2 deletions(-) create mode 100644 internal/api/handler/jobs.go create mode 100644 terraform/provider/internal/resources/job_resource.go diff --git a/deployments/k8s/operator.yaml b/deployments/k8s/operator.yaml index fa1fddb..3112d3b 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.3 в namespace sless +# - Deployment: оператор naeel/sless-operator:v0.1.4 в 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.3 + image: naeel/sless-operator:v0.1.4 # Always — чтобы всегда тянуть по точному тегу (не кешировать старый) imagePullPolicy: Always ports: diff --git a/internal/api/handler/jobs.go b/internal/api/handler/jobs.go new file mode 100644 index 0000000..7b5e3da --- /dev/null +++ b/internal/api/handler/jobs.go @@ -0,0 +1,149 @@ +// Изменено: 2026-03-07 +// jobs.go — CRUD handlers для FunctionJob CRD. +// Создаёт/читает/удаляет k8s FunctionJob ресурсы. +// Namespace берётся из URL: /v1/namespaces/{namespace}/jobs/{name} + +package handler + +import ( + "encoding/json" + "net/http" + + "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" +) + +// jobRequest — тело POST /v1/namespaces/{ns}/jobs +type jobRequest struct { + Name string `json:"name"` + FunctionRef string `json:"function"` + EventJSON string `json:"event_json,omitempty"` +} + +// jobResponse — ответ при чтении / создании FunctionJob +type jobResponse struct { + Name string `json:"name"` + Namespace string `json:"namespace"` + FunctionRef string `json:"function"` + EventJSON string `json:"event_json"` + Phase string `json:"phase"` + JobName string `json:"job_name,omitempty"` + StartTime string `json:"start_time,omitempty"` + CompletionTime string `json:"completion_time,omitempty"` + Message string `json:"message,omitempty"` +} + +// jobToResponse конвертирует FunctionJob CR → jobResponse. +func jobToResponse(j *slessv1alpha1.FunctionJob) jobResponse { + r := jobResponse{ + Name: j.Name, + Namespace: j.Namespace, + FunctionRef: j.Spec.FunctionRef, + EventJSON: j.Spec.EventJSON, + Phase: string(j.Status.Phase), + JobName: j.Status.JobName, + Message: j.Status.Message, + } + if j.Status.StartTime != nil { + r.StartTime = j.Status.StartTime.UTC().Format("2006-01-02T15:04:05Z") + } + if j.Status.CompletionTime != nil { + r.CompletionTime = j.Status.CompletionTime.UTC().Format("2006-01-02T15:04:05Z") + } + return r +} + +// CreateJob — POST /v1/namespaces/{namespace}/jobs +// Создаёт FunctionJob CR. Оператор запустит k8s Job асинхронно. +func (h *Handler) CreateJob(w http.ResponseWriter, r *http.Request) { + ns := namespace(r) + + var req jobRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errResp("invalid JSON: "+err.Error())) + return + } + if req.Name == "" { + writeJSON(w, http.StatusBadRequest, errResp("name is required")) + return + } + if req.FunctionRef == "" { + writeJSON(w, http.StatusBadRequest, errResp("function is required")) + return + } + if req.EventJSON == "" { + req.EventJSON = "{}" + } + + job := &slessv1alpha1.FunctionJob{ + ObjectMeta: metav1.ObjectMeta{ + Name: req.Name, + Namespace: ns, + }, + Spec: slessv1alpha1.FunctionJobSpec{ + FunctionRef: req.FunctionRef, + EventJSON: req.EventJSON, + }, + } + + if err := h.K8s.Create(r.Context(), job); err != nil { + if errors.IsAlreadyExists(err) { + writeJSON(w, http.StatusConflict, errResp("job already exists")) + return + } + h.Log.Error("create FunctionJob", "err", err) + writeJSON(w, http.StatusInternalServerError, errResp(err.Error())) + return + } + + writeJSON(w, http.StatusCreated, jobToResponse(job)) +} + +// GetJob — GET /v1/namespaces/{namespace}/jobs/{name} +// Возвращает статус FunctionJob включая phase и время выполнения. +func (h *Handler) GetJob(w http.ResponseWriter, r *http.Request) { + ns := namespace(r) + name := pathVar(r, "name") + + var job slessv1alpha1.FunctionJob + if err := h.K8s.Get(r.Context(), client.ObjectKey{Namespace: ns, Name: name}, &job); err != nil { + if errors.IsNotFound(err) { + writeJSON(w, http.StatusNotFound, errResp("job not found")) + return + } + h.Log.Error("get FunctionJob", "err", err) + writeJSON(w, http.StatusInternalServerError, errResp(err.Error())) + return + } + + writeJSON(w, http.StatusOK, jobToResponse(&job)) +} + +// DeleteJob — DELETE /v1/namespaces/{namespace}/jobs/{name} +// Удаляет FunctionJob CR (и подчинённый k8s Job через ownerReference, если ещё не убран по TTL). +func (h *Handler) DeleteJob(w http.ResponseWriter, r *http.Request) { + ns := namespace(r) + name := pathVar(r, "name") + + var job slessv1alpha1.FunctionJob + if err := h.K8s.Get(r.Context(), client.ObjectKey{Namespace: ns, Name: name}, &job); err != nil { + if errors.IsNotFound(err) { + w.WriteHeader(http.StatusNoContent) + return + } + h.Log.Error("get FunctionJob for delete", "err", err) + writeJSON(w, http.StatusInternalServerError, errResp(err.Error())) + return + } + + if err := h.K8s.Delete(r.Context(), &job); err != nil { + h.Log.Error("delete FunctionJob", "err", err) + writeJSON(w, http.StatusInternalServerError, errResp(err.Error())) + return + } + + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/api/router.go b/internal/api/router.go index f6876ee..70ebfa2 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -42,6 +42,11 @@ func NewRouter(h *handler.Handler, apiToken string, log *slog.Logger) http.Handl v1.HandleFunc("/namespaces/{namespace}/triggers/{name}", h.GetTrigger).Methods(http.MethodGet) v1.HandleFunc("/namespaces/{namespace}/triggers/{name}", h.DeleteTrigger).Methods(http.MethodDelete) + // Jobs CRUD — одноразовые запуски функций + v1.HandleFunc("/namespaces/{namespace}/jobs", h.CreateJob).Methods(http.MethodPost) + v1.HandleFunc("/namespaces/{namespace}/jobs/{name}", h.GetJob).Methods(http.MethodGet) + v1.HandleFunc("/namespaces/{namespace}/jobs/{name}", h.DeleteJob).Methods(http.MethodDelete) + // Цепочка middleware: logging → auth → router // Порядок важен: сначала логируем (чтобы видеть все запросы включая отклонённые), // затем проверяем авторизацию. diff --git a/terraform/provider/internal/client/client.go b/terraform/provider/internal/client/client.go index ca55673..0cd07dc 100644 --- a/terraform/provider/internal/client/client.go +++ b/terraform/provider/internal/client/client.go @@ -290,3 +290,102 @@ func (c *Client) DeleteTrigger(ctx context.Context, ns, name string) error { } return nil } + +// --- Job CRUD --- + +// JobRequest — тело POST /v1/namespaces/{ns}/jobs +type JobRequest struct { + Name string `json:"name"` + FunctionRef string `json:"function"` + EventJSON string `json:"event_json,omitempty"` +} + +// JobResponse — ответ GET /v1/namespaces/{ns}/jobs/{name} +type JobResponse struct { + Name string `json:"name"` + Namespace string `json:"namespace"` + FunctionRef string `json:"function"` + EventJSON string `json:"event_json"` + Phase string `json:"phase"` + JobName string `json:"job_name"` + StartTime string `json:"start_time"` + CompletionTime string `json:"completion_time"` + Message string `json:"message"` +} + +// CreateJob — POST /v1/namespaces/{ns}/jobs → 201 +func (c *Client) CreateJob(ctx context.Context, ns string, req JobRequest) (*JobResponse, error) { + url := fmt.Sprintf("%s/v1/namespaces/%s/jobs", c.endpoint, ns) + resp, err := c.doJSON(ctx, http.MethodPost, url, req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("create job: status %d: %s", resp.StatusCode, body) + } + var j JobResponse + return &j, json.NewDecoder(resp.Body).Decode(&j) +} + +// GetJob — GET /v1/namespaces/{ns}/jobs/{name} → nil если 404 +func (c *Client) GetJob(ctx context.Context, ns, name string) (*JobResponse, error) { + url := fmt.Sprintf("%s/v1/namespaces/%s/jobs/%s", c.endpoint, ns, name) + resp, err := c.doJSON(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound { + return nil, nil + } + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("get job: status %d: %s", resp.StatusCode, body) + } + var j JobResponse + return &j, json.NewDecoder(resp.Body).Decode(&j) +} + +// DeleteJob — DELETE /v1/namespaces/{ns}/jobs/{name} → 204 +func (c *Client) DeleteJob(ctx context.Context, ns, name string) error { + url := fmt.Sprintf("%s/v1/namespaces/%s/jobs/%s", c.endpoint, ns, name) + resp, err := c.doJSON(ctx, http.MethodDelete, url, nil) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotFound { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("delete job: status %d: %s", resp.StatusCode, body) + } + return nil +} + +// WaitJobDone опрашивает job каждые 5 секунд пока phase не Succeeded или Failed. +// Блокирует terraform apply до завершения джоба. +func (c *Client) WaitJobDone(ctx context.Context, ns, name string, timeout time.Duration) (*JobResponse, error) { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + j, err := c.GetJob(ctx, ns, name) + if err != nil { + return nil, err + } + if j == nil { + return nil, fmt.Errorf("job %s/%s not found while waiting", ns, name) + } + switch j.Phase { + case "Succeeded": + return j, nil + case "Failed": + return nil, fmt.Errorf("job failed: %s", j.Message) + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(5 * time.Second): + } + } + return nil, fmt.Errorf("timeout waiting for job %s/%s to complete", ns, name) +} diff --git a/terraform/provider/internal/provider/provider.go b/terraform/provider/internal/provider/provider.go index f6703ef..3528966 100644 --- a/terraform/provider/internal/provider/provider.go +++ b/terraform/provider/internal/provider/provider.go @@ -96,6 +96,7 @@ func (p *SlessProvider) Resources(_ context.Context) []func() resource.Resource return []func() resource.Resource{ resources.NewFunctionResource, resources.NewTriggerResource, + resources.NewJobResource, } } diff --git a/terraform/provider/internal/resources/job_resource.go b/terraform/provider/internal/resources/job_resource.go new file mode 100644 index 0000000..fee5df9 --- /dev/null +++ b/terraform/provider/internal/resources/job_resource.go @@ -0,0 +1,211 @@ +// 2026-03-07 +// job_resource.go — Terraform ресурс sless_job. +// +// Lifecycle: +// +// Create: POST /v1/namespaces/{ns}/jobs → WaitJobDone (10 мин) +// Блокирует terraform apply до завершения джоба (Succeeded/Failed). +// Если Failed — terraform apply падает с ошибкой. +// Read: GET /v1/namespaces/{ns}/jobs/{name} → sync phase/timing в state +// Delete: DELETE /v1/namespaces/{ns}/jobs/{name} +// Семантически no-op (джоб уже выполнен), но убирает CR из кластера. +// +// Update не поддерживается — любое изменение name/function/event_json требует пересоздания. +// Это корректно: job — одноразовое действие, нельзя "обновить" уже выполненное. +package resources + +import ( + "context" + "fmt" + "time" + + "terraform-provider-sless/internal/client" + + "github.com/hashicorp/terraform-plugin-framework/resource" + "github.com/hashicorp/terraform-plugin-framework/resource/schema" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier" + "github.com/hashicorp/terraform-plugin-framework/types" +) + +// jobTimeout — максимальное время ожидания завершения джоба. +const jobTimeout = 10 * time.Minute + +var _ resource.Resource = &JobResource{} + +type JobResource struct { + client *client.Client +} + +func NewJobResource() resource.Resource { + return &JobResource{} +} + +// JobModel — модель состояния terraform для sless_job. +type JobModel struct { + Namespace types.String `tfsdk:"namespace"` + Name types.String `tfsdk:"name"` + Function types.String `tfsdk:"function"` + EventJSON types.String `tfsdk:"event_json"` + Phase types.String `tfsdk:"phase"` + StartTime types.String `tfsdk:"start_time"` + CompletionTime types.String `tfsdk:"completion_time"` + Message types.String `tfsdk:"message"` +} + +func (r *JobResource) Metadata(_ context.Context, req resource.MetadataRequest, resp *resource.MetadataResponse) { + resp.TypeName = req.ProviderTypeName + "_job" +} + +func (r *JobResource) Schema(_ context.Context, _ resource.SchemaRequest, resp *resource.SchemaResponse) { + resp.Schema = schema.Schema{ + MarkdownDescription: "Одноразовый запуск serverless функции. terraform apply блокируется до завершения джоба.", + Attributes: map[string]schema.Attribute{ + // Все input-поля immutable — джоб нельзя "изменить", только пересоздать. + "namespace": schema.StringAttribute{ + Required: true, + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "name": schema.StringAttribute{ + Required: true, + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "function": schema.StringAttribute{ + Required: true, + MarkdownDescription: "Имя sless_function ресурса в том же namespace.", + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "event_json": schema.StringAttribute{ + Optional: true, + MarkdownDescription: `JSON-объект передаваемый в handle(event). По умолчанию "{}".`, + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + // Computed — заполняются после завершения джоба + "phase": schema.StringAttribute{ + Computed: true, + MarkdownDescription: "Фаза выполнения: Pending, Running, Succeeded, Failed.", + }, + "start_time": schema.StringAttribute{ + Computed: true, + MarkdownDescription: "Время запуска k8s Job (RFC3339).", + }, + "completion_time": schema.StringAttribute{ + Computed: true, + MarkdownDescription: "Время завершения k8s Job (RFC3339).", + }, + "message": schema.StringAttribute{ + Computed: true, + MarkdownDescription: "Результат выполнения или сообщение об ошибке.", + }, + }, + } +} + +func (r *JobResource) Configure(_ context.Context, req resource.ConfigureRequest, resp *resource.ConfigureResponse) { + if req.ProviderData == nil { + return + } + c, ok := req.ProviderData.(*client.Client) + if !ok { + resp.Diagnostics.AddError( + "unexpected provider data", + fmt.Sprintf("expected *client.Client, got: %T", req.ProviderData), + ) + return + } + r.client = c +} + +func (r *JobResource) Create(ctx context.Context, req resource.CreateRequest, resp *resource.CreateResponse) { + var plan JobModel + resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...) + if resp.Diagnostics.HasError() { + return + } + + ns := plan.Namespace.ValueString() + eventJSON := plan.EventJSON.ValueString() + if eventJSON == "" { + eventJSON = "{}" + } + + _, err := r.client.CreateJob(ctx, ns, client.JobRequest{ + Name: plan.Name.ValueString(), + FunctionRef: plan.Function.ValueString(), + EventJSON: eventJSON, + }) + if err != nil { + resp.Diagnostics.AddError("create job", err.Error()) + return + } + + // Блокируем apply до завершения джоба (Succeeded или Failed) + j, err := r.client.WaitJobDone(ctx, ns, plan.Name.ValueString(), jobTimeout) + if err != nil { + resp.Diagnostics.AddError("waiting for job to complete", err.Error()) + return + } + + resp.Diagnostics.Append(resp.State.Set(ctx, jobToModel(plan, j))...) +} + +func (r *JobResource) Read(ctx context.Context, req resource.ReadRequest, resp *resource.ReadResponse) { + var state JobModel + resp.Diagnostics.Append(req.State.Get(ctx, &state)...) + if resp.Diagnostics.HasError() { + return + } + + j, err := r.client.GetJob(ctx, state.Namespace.ValueString(), state.Name.ValueString()) + if err != nil { + resp.Diagnostics.AddError("read job", err.Error()) + return + } + if j == nil { + // Джоб удалён вне terraform — убираем из state + resp.State.RemoveResource(ctx) + return + } + + resp.Diagnostics.Append(resp.State.Set(ctx, jobToModel(state, j))...) +} + +// Update не реализован — все поля поддерживают только RequiresReplace. +// terraform-plugin-framework никогда не вызовет Update для этого ресурса. +func (r *JobResource) Update(_ context.Context, _ resource.UpdateRequest, resp *resource.UpdateResponse) { + resp.Diagnostics.AddError("update not supported", "sless_job does not support in-place updates") +} + +func (r *JobResource) Delete(ctx context.Context, req resource.DeleteRequest, resp *resource.DeleteResponse) { + var state JobModel + resp.Diagnostics.Append(req.State.Get(ctx, &state)...) + if resp.Diagnostics.HasError() { + return + } + + if err := r.client.DeleteJob(ctx, state.Namespace.ValueString(), state.Name.ValueString()); err != nil { + resp.Diagnostics.AddError("delete job", err.Error()) + } +} + +// jobToModel конвертирует API-ответ → state модель. +func jobToModel(plan JobModel, j *client.JobResponse) JobModel { + return JobModel{ + Namespace: types.StringValue(j.Namespace), + Name: types.StringValue(j.Name), + Function: types.StringValue(j.FunctionRef), + EventJSON: plan.EventJSON, // API отдаёт event_json но берём из plan для консистентности + Phase: types.StringValue(j.Phase), + StartTime: types.StringValue(j.StartTime), + CompletionTime: types.StringValue(j.CompletionTime), + Message: types.StringValue(j.Message), + } +}