Files
sless/terraform/provider/internal/resources/job_resource.go
T

269 lines
11 KiB
Go

// 2026-03-08
// job_resource.go — Terraform ресурс sless_job.
//
// Lifecycle:
//
// Create: POST /v1/namespaces/{ns}/jobs → WaitJobDone (10 мин)
// Блокирует terraform apply до завершения джоба (Succeeded/Failed).
// Если Failed — terraform apply падает с ошибкой.
// Если run_id=0 — джоб создаётся в k8s, но k8s Job не запускается.
// Read: GET /v1/namespaces/{ns}/jobs/{name} → sync phase/timing в state
// Delete: DELETE /v1/namespaces/{ns}/jobs/{name}
// Семантически no-op (джоб уже выполнен), но убирает CR из кластера.
//
// run_id: значение 0 = создать без запуска. >0 = запустить/перезапустить джоб.
// run_id имеет RequiresReplace: изменение значения (1→2→3) триггерирует повторный запуск.
package resources
import (
"context"
"fmt"
"time"
"terraform-provider-sless/internal/client"
"github.com/hashicorp/terraform-plugin-framework-validators/int64validator"
"github.com/hashicorp/terraform-plugin-framework/resource"
"github.com/hashicorp/terraform-plugin-framework/resource/schema"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/int64default"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/int64planmodifier"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier"
"github.com/hashicorp/terraform-plugin-framework/schema/validator"
"github.com/hashicorp/terraform-plugin-framework/types"
)
// defaultWaitTimeoutSec — дефолтный таймаут ожидания завершения джоба (600 сек = 10 мин).
// Пользователь может переопределить через wait_timeout_sec.
const defaultWaitTimeoutSec = 600
var _ resource.Resource = &JobResource{}
type JobResource struct {
client *client.Client
}
func NewJobResource() resource.Resource {
return &JobResource{}
}
// JobModel — модель состояния terraform для sless_job.
type JobModel struct {
Name types.String `tfsdk:"name"`
Function types.String `tfsdk:"function"`
EventJSON types.String `tfsdk:"event_json"`
// RunID: 0 = не запускать (Skipped), 1+ = запустить/перезапустить.
// RequiresReplace: изменение = пересоздание FunctionJob → новый запуск.
RunID types.Int64 `tfsdk:"run_id"`
// wait_timeout_sec — максимальное ожидание завершения джоба. Дефолт 600 сек.
WaitTimeoutSec types.Int64 `tfsdk:"wait_timeout_sec"`
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 — джоб нельзя "изменить", только пересоздать.
"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(),
},
}, // run_id: 0 = создать без запуска (Skipped), >0 = запустить.
// Изменение run_id (1→2→3...) триггерирует пересоздание = новый запуск.
"run_id": schema.Int64Attribute{
Optional: true,
Computed: true,
Default: int64default.StaticInt64(0),
MarkdownDescription: "0 = не запускать; >0 = запустить. Увеличьте run_id для повторного запуска джоба.",
PlanModifiers: []planmodifier.Int64{
int64planmodifier.RequiresReplace(),
},
Validators: []validator.Int64{
int64validator.AtLeast(0),
},
}, // wait_timeout_sec — сколько ждать завершения джоба. Увеличь если код долго работает (например миграция БД).
"wait_timeout_sec": schema.Int64Attribute{
Optional: true,
Computed: true,
MarkdownDescription: "Таймаут ожидания завершения джоба в секундах. По умолчанию 600 (10 мин).",
},
// 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 := r.client.Namespace
eventJSON := plan.EventJSON.ValueString()
if eventJSON == "" {
eventJSON = "{}"
}
// run_id=0: создаём FunctionJob в k8s, но оператор не запустит k8s Job.
// Пользователь может поменять run_id > 0 позже чтобы запустить.
runID := plan.RunID.ValueInt64()
_, err := r.client.CreateJob(ctx, ns, client.JobRequest{
Name: plan.Name.ValueString(),
FunctionRef: plan.Function.ValueString(),
EventJSON: eventJSON,
RunID: runID,
})
if err != nil {
resp.Diagnostics.AddError("create job", err.Error())
return
}
// Если RunID=0 — не ждём завершения, пишем state сразу
if runID == 0 {
// state: Namespace/Name/Function из plan, RunID=0, Phase=Skipped, остальное empty
waitTimeoutSec := plan.WaitTimeoutSec
if waitTimeoutSec.IsNull() || waitTimeoutSec.IsUnknown() || waitTimeoutSec.ValueInt64() <= 0 {
waitTimeoutSec = types.Int64Value(defaultWaitTimeoutSec)
}
resp.Diagnostics.Append(resp.State.Set(ctx, JobModel{
Name: types.StringValue(plan.Name.ValueString()),
Function: types.StringValue(plan.Function.ValueString()),
EventJSON: plan.EventJSON,
RunID: types.Int64Value(0),
WaitTimeoutSec: waitTimeoutSec,
Phase: types.StringValue("Skipped"),
StartTime: types.StringValue(""),
CompletionTime: types.StringValue(""),
Message: types.StringValue("run_id=0: set run_id>0 to execute"),
})...)
return
}
// Блокируем apply до завершения джоба (Succeeded или Failed)
waitSec := plan.WaitTimeoutSec.ValueInt64()
if waitSec <= 0 {
waitSec = defaultWaitTimeoutSec
}
j, err := r.client.WaitJobDone(ctx, ns, plan.Name.ValueString(), time.Duration(waitSec)*time.Second)
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, r.client.Namespace, 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, r.client.Namespace, state.Name.ValueString()); err != nil {
resp.Diagnostics.AddError("delete job", err.Error())
}
}
// jobToModel конвертирует API-ответ → state модель.
func jobToModel(plan JobModel, j *client.JobResponse) JobModel {
waitTimeoutSec := plan.WaitTimeoutSec
if waitTimeoutSec.IsNull() || waitTimeoutSec.IsUnknown() || waitTimeoutSec.ValueInt64() <= 0 {
waitTimeoutSec = types.Int64Value(defaultWaitTimeoutSec)
}
return JobModel{
Name: types.StringValue(j.Name),
Function: types.StringValue(j.FunctionRef),
EventJSON: plan.EventJSON,
RunID: types.Int64Value(j.RunID),
WaitTimeoutSec: waitTimeoutSec,
Phase: types.StringValue(j.Phase),
StartTime: types.StringValue(j.StartTime),
CompletionTime: types.StringValue(j.CompletionTime),
Message: types.StringValue(j.Message),
}
}