A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
14 kB · 376 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377package controller
import ( "context" "errors" "fmt" "io" "net/http" "os" "path/filepath" "strings" "time"
"connectrpc.com/connect"
controllerv1 "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/controller/v1" appnotify "forgejo.alexma.top/alexma233/composia/internal/app/notify" "forgejo.alexma.top/alexma233/composia/internal/core/config" corenotify "forgejo.alexma.top/alexma233/composia/internal/core/notify" "forgejo.alexma.top/alexma233/composia/internal/core/task" "forgejo.alexma.top/alexma233/composia/internal/platform/rpcutil" "forgejo.alexma.top/alexma233/composia/internal/platform/store")
type taskServer struct { db *store.DB cfg *config.ControllerConfig availableNodeIDs map[string]struct{} taskQueue *taskQueueNotifier taskResults *taskResultNotifier notifier *appnotify.Notifier}
const ( confirmationDecisionApprove = "approve" confirmationDecisionReject = "reject")
func (server *taskServer) ListTasks(ctx context.Context, req *connect.Request[controllerv1.ListTasksRequest]) (*connect.Response[controllerv1.ListTasksResponse], error) { if req.Msg == nil { req.Msg = &controllerv1.ListTasksRequest{} }
tasks, totalCount, err := server.db.ListTasks(ctx, taskStatusesFromProto(req.Msg.GetStatus()), req.Msg.GetServiceName(), req.Msg.GetNodeId(), taskTypesFromProto(req.Msg.GetType()), taskStatusesFromProto(req.Msg.GetExcludeStatus()), req.Msg.GetExcludeServiceName(), req.Msg.GetExcludeNodeId(), taskTypesFromProto(req.Msg.GetExcludeType()), req.Msg.GetPage(), req.Msg.GetPageSize()) if err != nil { return nil, connect.NewError(connect.CodeInternal, err) }
response := &controllerv1.ListTasksResponse{ Tasks: make([]*controllerv1.TaskSummary, 0, len(tasks)), TotalCount: totalCount, } for _, record := range tasks { response.Tasks = append(response.Tasks, taskSummaryMessage(record)) }
return connect.NewResponse(response), nil}
func taskSummaryMessage(record store.TaskSummary) *controllerv1.TaskSummary { createdAt, _ := time.Parse(time.RFC3339, record.CreatedAt) return &controllerv1.TaskSummary{ TaskId: record.TaskID, Type: protoTaskType(task.Type(record.Type)), Status: protoTaskStatus(task.Status(record.Status)), ServiceName: record.ServiceName, NodeId: record.NodeID, CreatedAt: protoTaskTime(createdAt), }}
func (server *taskServer) GetTask(ctx context.Context, req *connect.Request[controllerv1.GetTaskRequest]) (*connect.Response[controllerv1.GetTaskResponse], error) { if req.Msg == nil || req.Msg.GetTaskId() == "" { return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("task_id is required")) }
detail, err := server.db.GetTask(ctx, req.Msg.GetTaskId()) if err != nil { if errors.Is(err, store.ErrTaskNotFound) { return nil, connect.NewError(connect.CodeNotFound, err) } return nil, connect.NewError(connect.CodeInternal, err) }
response := &controllerv1.GetTaskResponse{ TaskId: detail.Record.TaskID, Type: protoTaskType(detail.Record.Type), Source: protoTaskSource(detail.Record.Source), ServiceName: detail.Record.ServiceName, NodeId: detail.Record.NodeID, Status: protoTaskStatus(detail.Record.Status), CreatedAt: protoTaskTime(detail.Record.CreatedAt), StartedAt: protoNullableTaskTime(detail.Record.StartedAt), FinishedAt: protoNullableTaskTime(detail.Record.FinishedAt), RepoRevision: detail.Record.RepoRevision, ErrorSummary: detail.Record.ErrorSummary, TriggeredBy: detail.Record.TriggeredBy, ResultRevision: detail.Record.ResultRevision, AttemptOfTaskId: detail.Record.AttemptOfTaskID, ExecutionState: string(detail.Record.ExecutionState), LeaseExpiresAt: protoNullableTaskTime(detail.Record.LeaseExpiresAt), Steps: make([]*controllerv1.TaskStepSummary, 0, len(detail.Steps)), } for _, step := range detail.Steps { response.Steps = append(response.Steps, &controllerv1.TaskStepSummary{ StepName: protoTaskStepName(step.StepName), Status: protoTaskStatus(step.Status), StartedAt: protoNullableTaskTime(step.StartedAt), FinishedAt: protoNullableTaskTime(step.FinishedAt), }) }
return connect.NewResponse(response), nil}
func (server *taskServer) FailLostTaskExecution(ctx context.Context, req *connect.Request[controllerv1.FailLostTaskExecutionRequest]) (*connect.Response[controllerv1.TaskActionResponse], error) { if req.Msg == nil || req.Msg.GetTaskId() == "" { return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("task_id is required")) } detail, err := server.db.GetTask(ctx, req.Msg.GetTaskId()) if err != nil { return nil, connect.NewError(connect.CodeNotFound, err) } if detail.Record.Status != task.StatusRunning || detail.Record.ExecutionState != task.ExecutionLeaseLost { return nil, connect.NewError(connect.CodeFailedPrecondition, errors.New("task execution lease is not lost")) } summary := strings.TrimSpace(req.Msg.GetErrorSummary()) if summary == "" { summary = "task execution lease was lost; outcome requires reconciliation" } finishedAt := time.Now().UTC() if err := server.db.FailLostTaskExecution(ctx, detail.Record.TaskID, finishedAt, summary); err != nil { if errors.Is(err, store.ErrTaskExecutionMismatch) { return nil, connect.NewError(connect.CodeFailedPrecondition, err) } return nil, connect.NewError(connect.CodeInternal, err) } detail.Record.Status = task.StatusFailed detail.Record.ErrorSummary = summary detail.Record.FinishedAt = &finishedAt notifyTaskResult(server.taskResults, detail.Record.TaskID) return connect.NewResponse(taskActionResponse(detail.Record)), nil}
func (server *taskServer) TailTaskLogs(ctx context.Context, req *connect.Request[controllerv1.TailTaskLogsRequest], stream *connect.ServerStream[controllerv1.TailTaskLogsResponse]) error { if req.Msg == nil || req.Msg.GetTaskId() == "" { return connect.NewError(connect.CodeInvalidArgument, errors.New("task_id is required")) }
detail, err := server.db.GetTask(ctx, req.Msg.GetTaskId()) if err != nil { if errors.Is(err, store.ErrTaskNotFound) { return connect.NewError(connect.CodeNotFound, err) } return connect.NewError(connect.CodeInternal, err) } if detail.Record.LogPath == "" { return connect.NewError(connect.CodeFailedPrecondition, errors.New("task does not have a log file")) }
var offset int64 ticker := time.NewTicker(200 * time.Millisecond) defer ticker.Stop()
for { content, nextOffset, err := readNewLogContent(detail.Record.LogPath, offset) if err != nil { return connect.NewError(connect.CodeInternal, err) } offset = nextOffset if content != "" { if err := stream.Send(&controllerv1.TailTaskLogsResponse{Content: content}); err != nil { return err } } refreshedDetail, err := server.db.GetTask(ctx, req.Msg.GetTaskId()) if err != nil { if errors.Is(err, store.ErrTaskNotFound) { return connect.NewError(connect.CodeNotFound, err) } return connect.NewError(connect.CodeInternal, err) } detail = refreshedDetail if content == "" && isTerminalTaskStatus(detail.Record.Status) { return nil }
select { case <-ctx.Done(): return nil case <-ticker.C: } }}
func (server *taskServer) RunTaskAgain(ctx context.Context, req *connect.Request[controllerv1.RunTaskAgainRequest]) (*connect.Response[controllerv1.TaskActionResponse], error) { if req.Msg == nil || req.Msg.GetTaskId() == "" { return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("task_id is required")) } detail, err := server.db.GetTask(ctx, req.Msg.GetTaskId()) if err != nil { if errors.Is(err, store.ErrTaskNotFound) { return nil, connect.NewError(connect.CodeNotFound, err) } return nil, connect.NewError(connect.CodeInternal, err) } if detail.Record.ServiceName == "" { return nil, connect.NewError(connect.CodeFailedPrecondition, errors.New("task cannot be rerun without service_name")) }
var rerunType task.Type switch detail.Record.Type { case task.TypeDeploy, task.TypeUpdate, task.TypeStop, task.TypeRestart, task.TypeBackup: rerunType = detail.Record.Type default: return nil, connect.NewError(connect.CodeFailedPrecondition, fmt.Errorf("task type %q cannot be rerun yet", detail.Record.Type)) }
params, err := taskParams(detail.Record.ParamsJSON) if err != nil { return nil, connect.NewError(connect.CodeInternal, err) } var targetNodeIDs []string if detail.Record.NodeID != "" { targetNodeIDs = []string{detail.Record.NodeID} } createdTask, err := createServiceTaskWithOptions(ctx, server.db, server.cfg, server.availableNodeIDs, detail.Record.ServiceName, targetNodeIDs, rerunType, params.DataNames, serviceTaskCreateOptions{AttemptOfTaskID: detail.Record.TaskID, Source: requestTaskSource(req.Header()), ComposeRecreateMode: params.ComposeRecreateMode}) if err != nil { return nil, err } return connect.NewResponse(taskActionResponse(createdTask)), nil}
func (server *taskServer) ResolveTaskConfirmation(ctx context.Context, req *connect.Request[controllerv1.ResolveTaskConfirmationRequest]) (*connect.Response[controllerv1.TaskActionResponse], error) { if req.Msg == nil || req.Msg.GetTaskId() == "" { return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("task_id is required")) } decision := taskConfirmationDecisionFromProto(req.Msg.GetDecision()) if decision != confirmationDecisionApprove && decision != confirmationDecisionReject { return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("decision must be approve or reject")) }
detail, err := server.db.GetTask(ctx, req.Msg.GetTaskId()) if err != nil { if errors.Is(err, store.ErrTaskNotFound) { return nil, connect.NewError(connect.CodeNotFound, err) } return nil, connect.NewError(connect.CodeInternal, err) } if detail.Record.Type != task.TypeMigrate { return nil, connect.NewError(connect.CodeFailedPrecondition, fmt.Errorf("task type %q does not support confirmation resolution", detail.Record.Type)) } if detail.Record.Status != task.StatusAwaitingConfirmation { return nil, connect.NewError(connect.CodeFailedPrecondition, fmt.Errorf("task %q is not awaiting confirmation", detail.Record.TaskID)) }
comment := strings.TrimSpace(req.Msg.GetComment()) resolvedBy, _ := rpcutil.BearerSubject(ctx) resolvedLine := "manual confirmation decision=" + decision if resolvedBy != "" { resolvedLine += " actor=" + resolvedBy } if comment != "" { resolvedLine += fmt.Sprintf(" comment=%q", comment) } resolvedLine += "\n" if err := appendTaskLogRaw(detail.Record.LogPath, resolvedLine); err != nil { return nil, connect.NewError(connect.CodeInternal, err) }
finishedAt := time.Now().UTC() stepStatus := task.StatusSucceeded errorSummary := "" if decision == confirmationDecisionReject { stepStatus = task.StatusCancelled errorSummary = "manual verification rejected" if comment != "" { errorSummary = "manual verification rejected: " + comment } } if err := server.db.UpsertTaskStep(ctx, task.StepRecord{TaskID: detail.Record.TaskID, StepName: task.StepAwaitingConfirmation, Status: stepStatus, StartedAt: findTaskStepStartedAt(detail.Steps, task.StepAwaitingConfirmation), FinishedAt: &finishedAt}); err != nil { return nil, connect.NewError(connect.CodeInternal, err) }
if decision == confirmationDecisionApprove { if err := server.db.TransitionTaskStatus(ctx, detail.Record.TaskID, task.StatusAwaitingConfirmation, task.StatusPending, ""); err != nil { return nil, connect.NewError(connect.CodeInternal, err) } detail.Record.Status = task.StatusPending notifyTaskQueue(server.taskQueue) return connect.NewResponse(taskActionResponse(detail.Record)), nil }
if err := server.db.CompleteTask(ctx, detail.Record.TaskID, task.StatusCancelled, finishedAt, errorSummary); err != nil { return nil, connect.NewError(connect.CodeInternal, err) } detail.Record.Status = task.StatusCancelled detail.Record.FinishedAt = &finishedAt detail.Record.ErrorSummary = errorSummary notifyTaskResult(server.taskResults, detail.Record.TaskID) dispatchTaskRecordNotification(server.notifier, corenotify.EventTaskCancelled, detail.Record) return connect.NewResponse(taskActionResponse(detail.Record)), nil}
func taskActionResponse(record task.Record) *controllerv1.TaskActionResponse { return &controllerv1.TaskActionResponse{ TaskId: record.TaskID, Status: protoTaskStatus(record.Status), RepoRevision: record.RepoRevision, }}
func readNewLogContent(logPath string, offset int64) (string, int64, error) { file, err := os.Open(logPath) //nolint:gosec if err != nil { return "", offset, fmt.Errorf("open task log %q: %w", logPath, err) } defer func() { _ = file.Close() }()
stat, err := file.Stat() if err != nil { return "", offset, fmt.Errorf("stat task log %q: %w", logPath, err) } if offset > stat.Size() { offset = 0 } if _, err := file.Seek(offset, io.SeekStart); err != nil { return "", offset, fmt.Errorf("seek task log %q: %w", logPath, err) } content, err := io.ReadAll(file) if err != nil { return "", offset, fmt.Errorf("read task log %q: %w", logPath, err) } return string(content), stat.Size(), nil}
func isTerminalTaskStatus(status task.Status) bool { switch status { case task.StatusSucceeded, task.StatusFailed, task.StatusCancelled: return true default: return false }}
func requestTaskSource(header http.Header) task.Source { switch strings.ToLower(strings.TrimSpace(header.Get("X-Composia-Source"))) { case string(task.SourceWeb): return task.SourceWeb case string(task.SourceOthers): return task.SourceOthers case string(task.SourceSchedule): return task.SourceSchedule case string(task.SourceSystem): return task.SourceSystem default: return task.SourceCLI }}
func sourceHeader(source task.Source) http.Header { if source == "" { return make(http.Header) } header := make(http.Header) header.Set("X-Composia-Source", string(source)) return header}
func mustRelativeServiceDir(repoDir, serviceDir string) string { relativePath, err := filepath.Rel(repoDir, serviceDir) if err != nil { return filepath.ToSlash(serviceDir) } return filepath.ToSlash(relativePath)}