A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137package cli
import ( "context" "errors" "flag" "fmt" "time"
controllerv1 "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/controller/v1")
type waitOptions struct { wait *bool follow *bool timeout *time.Duration interval *time.Duration}
func addWaitFlags(fs *flag.FlagSet) waitOptions { wait := fs.Bool("wait", false, "wait for the task to finish") follow := fs.Bool("follow", false, "stream task logs while waiting") timeout := fs.Duration("timeout", 0, "maximum wait duration") interval := fs.Duration("interval", time.Second, "poll interval") return waitOptions{wait: wait, follow: follow, timeout: timeout, interval: interval}}
func (options waitOptions) shouldWait() bool { return (options.wait != nil && *options.wait) || (options.follow != nil && *options.follow)}
func (application *app) printTaskActionWithWait(response *controllerv1.TaskActionResponse, options waitOptions) error { if err := application.printTaskAction(response); err != nil { return err } if !options.shouldWait() { return nil } return application.waitTask(response.GetTaskId(), options)}
func (application *app) printServiceActionWithWait(response *controllerv1.RunServiceActionResponse, options waitOptions) error { if err := application.printServiceAction(response); err != nil { return err } if !options.shouldWait() { return nil } tasks := response.GetTasks() if len(tasks) == 0 { return errors.New("at least one task is required") } if options.follow != nil && *options.follow && len(tasks) != 1 { return errors.New("--follow requires exactly one queued task") } for _, queuedTask := range tasks { if err := application.waitTask(queuedTask.GetTaskId(), options); err != nil { return err } } return nil}
func (application *app) waitTask(taskID string, options waitOptions) error { if taskID == "" { return errors.New("task_id is required") } ctx := application.ctx if options.timeout != nil && *options.timeout > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, *options.timeout) defer cancel() } if options.follow != nil && *options.follow { if err := application.streamTaskLogs(ctx, taskID); err != nil { return err } } interval := time.Second if options.interval != nil && *options.interval > 0 { interval = *options.interval } ticker := time.NewTicker(interval) defer ticker.Stop() for { response, err := application.client.tasks.GetTask(ctx, newRequest(&controllerv1.GetTaskRequest{TaskId: taskID})) if err != nil { return err } status := response.Msg.GetStatus() if isTerminalStatus(status) { if !application.isJSONOutput() { if err := application.writeKV([][2]string{{"final_status", taskStatusText(status)}}); err != nil { return err } } if status != controllerv1.TaskStatus_TASK_STATUS_SUCCEEDED { return fmt.Errorf("task %s finished with status %s", taskID, taskStatusText(status)) } return nil } select { case <-ctx.Done(): return ctx.Err() case <-ticker.C: } }}
func (application *app) streamTaskLogs(ctx context.Context, taskID string) error { stream, err := application.client.tasks.TailTaskLogs(ctx, newRequest(&controllerv1.TailTaskLogsRequest{TaskId: taskID})) if err != nil { return err } for stream.Receive() { if application.isJSONOutput() { if err := application.printMessage(stream.Msg()); err != nil { return err } continue } if _, err := fmt.Fprint(application.out, stream.Msg().GetContent()); err != nil { return err } } return stream.Err()}
func isTerminalStatus(status controllerv1.TaskStatus) bool { switch status { case controllerv1.TaskStatus_TASK_STATUS_SUCCEEDED, controllerv1.TaskStatus_TASK_STATUS_FAILED, controllerv1.TaskStatus_TASK_STATUS_CANCELLED: return true default: return false }}