Kiu/Semaphore
Kiu/Semaphore
1
0
Fork 0
mirror of https://github.com/semaphoreui/semaphore.git synced 2025-01-20 15:29:28 +01:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

task stages (#2103)

Browse Source 
* feat(be): add stages

* feat(tf): split to plan/apply

* fix(be): blocking
This commit is contained in:
Denis Gukov 2024-06-17 23:37:45 +05:00 committed by GitHub
parent 40b8f659a7
commit 91601eb0eb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 241 additions and 222 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
api/projects/tasks.go
View File

@ -110,6 +110,23 @@ func GetTaskMiddleware(next http.Handler) http.Handler {
})
}
// GetTaskOutput returns the logged task output by id and writes it as json or returns error
func GetTaskStages(w http.ResponseWriter, r *http.Request) {
task := context.Get(r, "task").(db.Task)
project := context.Get(r, "project").(db.Project)
var output []db.TaskOutput
output, err := helpers.Store(r).GetTaskOutputs(project.ID, task.ID)
if err != nil {
util.LogErrorWithFields(err, log.Fields{"error": "Bad request. Cannot get task output from database"})
w.WriteHeader(http.StatusBadRequest)
return
}
helpers.WriteJSON(w, http.StatusOK, output)
}
// GetTaskOutput returns the logged task output by id and writes it as json or returns error
func GetTaskOutput(w http.ResponseWriter, r *http.Request) {
task := context.Get(r, "task").(db.Task)

7
db/Store.go
View File

@ -235,6 +235,8 @@ type Store interface {
DeleteTaskWithOutputs(projectID int, taskID int) error
GetTaskOutputs(projectID int, taskID int) ([]TaskOutput, error)
CreateTaskOutput(output TaskOutput) (TaskOutput, error)
GetTaskStages(projectID int, taskID int) ([]TaskStage, error)
CreateTaskStage(stage TaskStage) (TaskStage, error)
GetView(projectID int, viewID int) (View, error)
GetViews(projectID int) ([]View, error)
@ -381,6 +383,11 @@ var TaskOutputProps = ObjectProps{
Type: reflect.TypeOf(TaskOutput{}),
}
var TaskStageProps = ObjectProps{
TableName: "task__stage",
Type: reflect.TypeOf(TaskStage{}),
}
var ViewProps = ObjectProps{
TableName: "project__view",
Type: reflect.TypeOf(View{}),

17
db/Task.go
View File

@ -121,3 +121,20 @@ type TaskOutput struct {
Time time.Time `db:"time" json:"time"`
Output string `db:"output" json:"output"`
}
type TaskStageType string
const (
TaskStageRepositoryClone TaskStageType = "repository_clone"
TaskStageTerraformPlan TaskStageType = "terraform_plan"
TaskStageTerraformApply TaskStageType = "terraform_apply"
)
type TaskStage struct {
TaskID int `db:"task_id" json:"task_id"`
Start *time.Time `db:"start" json:"start"`
End *time.Time `db:"end" json:"end"`
StartOutputID *int `db:"start_output_id" json:"start_output_id"`
EndOutputID *int `db:"end_output_id" json:"end_output_id"`
Type TaskStageType `db:"type" json:"type"`
}

21
db/bolt/task.go
View File

@ -6,6 +6,27 @@ import (
"time"
)
func (d *BoltDb) CreateTaskStage(stage db.TaskStage) (db.TaskStage, error) {
newOutput, err := d.createObject(stage.TaskID, db.TaskStageProps, stage)
if err != nil {
return db.TaskStage{}, err
}
return newOutput.(db.TaskStage), nil
}
func (d *BoltDb) GetTaskStages(projectID int, taskID int) (res []db.TaskStage, err error) {
// check if task exists in the project
_, err = d.GetTask(projectID, taskID)
if err != nil {
return
}
err = d.getObjects(taskID, db.TaskStageProps, db.RetrieveQueryParams{}, nil, &res)
return
}
func (d *BoltDb) CreateTask(task db.Task) (newTask db.Task, err error) {
task.Created = time.Now()
res, err := d.createObject(0, db.TaskProps, task)

14
db/sql/migrations/v2.10.5.sql Normal file
View File

@ -0,0 +1,14 @@
create table task__stage(
`id` integer primary key autoincrement,
`task_id` int not null,
`start` datetime not null,
`start_output_id` int,
`end` datetime,
`end_output_id` int,
`type` varchar(20) not null,
`status` varchat(255) not null,
`result` text,
foreign key (`task_id`) references project(`id`),
foreign key (`start_output_id`) references task__output(`id`),
foreign key (`end_output_id`) references task__output(`id`) on delete cascade
);

15
db/sql/task.go
View File

@ -2,10 +2,23 @@ package sql
import (
"database/sql"
"github.com/ansible-semaphore/semaphore/db"
"github.com/Masterminds/squirrel"
"github.com/ansible-semaphore/semaphore/db"
)
func (d *SqlDb) CreateTaskStage(stage db.TaskStage) (db.TaskStage, error) {
_, err := d.exec(
"insert into task__stage (task_id, type) VALUES (?, ?, ?, ?)",
stage.TaskID,
stage.Type,
stage.Start)
return stage, err
}
func (d *SqlDb) GetTaskStages(projectID int, taskID int) ([]db.TaskStage, error) {
return nil, nil
}
func (d *SqlDb) CreateTask(task db.Task) (db.Task, error) {
err := d.sql.Insert(&task)
return task, err

29
db_lib/BashApp.go
View File

@ -8,13 +8,35 @@ import (
"os"
"os/exec"
"strings"
"time"
)
type BashApp struct {
Logger task_logger.Logger
//Playbook *AnsiblePlaybook
Template db.Template
Repository db.Repository
reader bashReader
}
type bashReader struct {
input *string
logger task_logger.Logger
}
func (r *bashReader) Read(p []byte) (n int, err error) {
r.logger.SetStatus(task_logger.TaskWaitingConfirmation)
for {
time.Sleep(time.Second * 3)
if r.input != nil {
break
}
}
copy(p, *r.input+"\n")
r.logger.SetStatus(task_logger.TaskRunningStatus)
return len(*r.input) + 1, nil
}
func (t *BashApp) makeCmd(command string, args []string, environmentVars *[]string) *exec.Cmd {
@ -54,6 +76,10 @@ func (t *BashApp) GetFullPath() (path string) {
func (t *BashApp) SetLogger(logger task_logger.Logger) task_logger.Logger {
t.Logger = logger
t.Logger.AddStatusListener(func(status task_logger.TaskStatus) {
})
t.reader.logger = logger
return logger
}
@ -64,6 +90,7 @@ func (t *BashApp) InstallRequirements() error {
func (t *BashApp) Run(args []string, environmentVars *[]string, inputs map[string]string, cb func(*os.Process)) error {
cmd := t.makeCmd("bash", args, environmentVars)
t.Logger.LogCmd(cmd)
//cmd.Stdin = &t.reader
cmd.Stdin = strings.NewReader("")
err := cmd.Start()
if err != nil {

146
db_lib/PulumiApp.go
View File

@ -1,146 +0,0 @@
package db_lib
import (
"fmt"
"github.com/ansible-semaphore/semaphore/db"
"github.com/ansible-semaphore/semaphore/pkg/task_logger"
"github.com/ansible-semaphore/semaphore/util"
"os"
"os/exec"
"path"
"strings"
"time"
)
type PulumiApp struct {
Logger task_logger.Logger
Template db.Template
Repository db.Repository
reader PulumiReader
}
type PulumiLogger struct {
logger task_logger.Logger
reader *PulumiReader
}
func (l *PulumiLogger) Log(msg string) {
l.logger.Log(msg)
}
func (l *PulumiLogger) Logf(format string, a ...any) {
l.logger.Logf(format, a...)
}
type PulumiReader struct {
confirmed bool
logger *PulumiLogger
}
func (r *PulumiReader) Read(p []byte) (n int, err error) {
if r.confirmed {
copy(p, "\n")
return 1, nil
}
r.logger.SetStatus(task_logger.TaskWaitingConfirmation)
for {
time.Sleep(time.Second * 3)
if r.confirmed {
break
}
}
copy(p, "yes\n")
r.logger.SetStatus(task_logger.TaskRunningStatus)
return 4, nil
}
func (l *PulumiLogger) LogWithTime(now time.Time, msg string) {
l.logger.LogWithTime(now, msg)
}
func (l *PulumiLogger) LogfWithTime(now time.Time, format string, a ...any) {
l.logger.LogWithTime(now, fmt.Sprintf(format, a...))
}
func (l *PulumiLogger) LogCmd(cmd *exec.Cmd) {
l.logger.LogCmd(cmd)
}
func (l *PulumiLogger) SetStatus(status task_logger.TaskStatus) {
if status == task_logger.TaskConfirmed {
l.reader.confirmed = true
}
l.logger.SetStatus(status)
}
func (t *PulumiApp) makeCmd(command string, args []string, environmentVars *[]string) *exec.Cmd {
cmd := exec.Command(command, args...) //nolint: gas
cmd.Dir = t.GetFullPath()
cmd.Env = os.Environ()
cmd.Env = append(cmd.Env, fmt.Sprintf("HOME=%s", util.Config.TmpPath))
cmd.Env = append(cmd.Env, fmt.Sprintf("PWD=%s", cmd.Dir))
if environmentVars != nil {
cmd.Env = append(cmd.Env, *environmentVars...)
}
// Remove sensitive env variables from cmd process
for _, env := range getSensitiveEnvs() {
cmd.Env = append(cmd.Env, env+"=")
}
return cmd
}
func (t *PulumiApp) runCmd(command string, args []string) error {
cmd := t.makeCmd(command, args, nil)
t.Logger.LogCmd(cmd)
return cmd.Run()
}
func (t *PulumiApp) GetFullPath() string {
return path.Join(t.Repository.GetFullPath(t.Template.ID), strings.TrimPrefix(t.Template.Playbook, "/"))
}
func (t *PulumiApp) SetLogger(logger task_logger.Logger) task_logger.Logger {
internalLogger := &PulumiLogger{
logger: logger,
reader: &t.reader,
}
t.reader.logger = internalLogger
t.Logger = internalLogger
return internalLogger
}
func (t *PulumiApp) InstallRequirements() error {
if _, ok := t.Logger.(*PulumiLogger); !ok {
t.SetLogger(t.Logger)
}
cmd := t.makeCmd("pulumi", []string{"stack", "init"}, nil)
t.Logger.LogCmd(cmd)
err := cmd.Start()
if err != nil {
return err
}
return cmd.Wait()
}
func (t *PulumiApp) Run(args []string, environmentVars *[]string, inputs map[string]string, cb func(*os.Process)) error {
cmd := t.makeCmd("pulumi", args, environmentVars)
t.Logger.LogCmd(cmd)
cmd.Stdin = &t.reader
err := cmd.Start()
if err != nil {
return err
}
cb(cmd.Process)
return cmd.Wait()
}

141
db_lib/TerraformApp.go
View File

@ -25,64 +25,18 @@ type TerraformApp struct {
Repository db.Repository
reader terraformReader
Name TerraformAppName
noChanges bool
}
type terraformLogger struct {
logger task_logger.Logger
reader *terraformReader
}
type terraformReaderResult int
func (l *terraformLogger) Log(msg string) {
l.logger.Log(msg)
}
func (l *terraformLogger) Logf(format string, a ...any) {
l.logger.Logf(format, a...)
}
const (
terraformReaderConfirmed terraformReaderResult = iota
terraformReaderFailed
)
type terraformReader struct {
confirmed bool
logger *terraformLogger
}
func (r *terraformReader) Read(p []byte) (n int, err error) {
if r.confirmed {
copy(p, "\n")
return 1, nil
}
r.logger.SetStatus(task_logger.TaskWaitingConfirmation)
for {
time.Sleep(time.Second * 3)
if r.confirmed {
break
}
}
copy(p, "yes\n")
r.logger.SetStatus(task_logger.TaskRunningStatus)
return 4, nil
}
func (l *terraformLogger) LogWithTime(now time.Time, msg string) {
l.logger.LogWithTime(now, msg)
}
func (l *terraformLogger) LogfWithTime(now time.Time, format string, a ...any) {
l.logger.LogWithTime(now, fmt.Sprintf(format, a...))
}
func (l *terraformLogger) LogCmd(cmd *exec.Cmd) {
l.logger.LogCmd(cmd)
}
func (l *terraformLogger) SetStatus(status task_logger.TaskStatus) {
if status == task_logger.TaskConfirmed {
l.reader.confirmed = true
}
l.logger.SetStatus(status)
result *terraformReaderResult
}
func (t *TerraformApp) makeCmd(command string, args []string, environmentVars *[]string) *exec.Cmd {
@ -116,22 +70,32 @@ func (t *TerraformApp) GetFullPath() string {
}
func (t *TerraformApp) SetLogger(logger task_logger.Logger) task_logger.Logger {
internalLogger := &terraformLogger{
logger: logger,
reader: &t.reader,
}
t.Logger = logger
t.reader.logger = internalLogger
t.Logger = internalLogger
return internalLogger
t.Logger.AddLogListener(func(new time.Time, msg string) {
if strings.Contains(msg, "No changes.") {
t.noChanges = true
}
})
t.Logger.AddStatusListener(func(status task_logger.TaskStatus) {
var result terraformReaderResult
switch status {
case task_logger.TaskConfirmed:
result = terraformReaderConfirmed
t.reader.result = &result
case task_logger.TaskFailStatus, task_logger.TaskStoppedStatus:
result = terraformReaderFailed
t.reader.result = &result
}
})
return logger
}
func (t *TerraformApp) InstallRequirements() error {
if _, ok := t.Logger.(*terraformLogger); !ok {
t.SetLogger(t.Logger)
}
cmd := t.makeCmd(string(t.Name), []string{"init"}, nil)
t.Logger.LogCmd(cmd)
err := cmd.Start()
@ -141,10 +105,11 @@ func (t *TerraformApp) InstallRequirements() error {
return cmd.Wait()
}
func (t *TerraformApp) Run(args []string, environmentVars *[]string, inputs map[string]string, cb func(*os.Process)) error {
func (t *TerraformApp) Plan(args []string, environmentVars *[]string, inputs map[string]string, cb func(*os.Process)) error {
args = append([]string{"plan"}, args...)
cmd := t.makeCmd(string(t.Name), args, environmentVars)
t.Logger.LogCmd(cmd)
cmd.Stdin = &t.reader
cmd.Stdin = strings.NewReader("")
err := cmd.Start()
if err != nil {
return err
@ -152,3 +117,47 @@ func (t *TerraformApp) Run(args []string, environmentVars *[]string, inputs map[
cb(cmd.Process)
return cmd.Wait()
}
func (t *TerraformApp) Apply(args []string, environmentVars *[]string, inputs map[string]string, cb func(*os.Process)) error {
args = append([]string{"apply", "-auto-approve"}, args...)
cmd := t.makeCmd(string(t.Name), args, environmentVars)
t.Logger.LogCmd(cmd)
cmd.Stdin = strings.NewReader("")
err := cmd.Start()
if err != nil {
return err
}
cb(cmd.Process)
return cmd.Wait()
}
func (t *TerraformApp) Run(args []string, environmentVars *[]string, inputs map[string]string, cb func(*os.Process)) error {
err := t.Plan(args, environmentVars, inputs, cb)
if err != nil {
return err
}
if t.noChanges {
t.Logger.SetStatus(task_logger.TaskSuccessStatus)
return nil
}
t.Logger.SetStatus(task_logger.TaskWaitingConfirmation)
for {
time.Sleep(time.Second * 3)
if t.reader.result != nil {
break
}
}
switch *t.reader.result {
case terraformReaderFailed:
return nil
case terraformReaderConfirmed:
t.Logger.SetStatus(task_logger.TaskRunningStatus)
return t.Apply(args, environmentVars, inputs, cb)
default:
return fmt.Errorf("unknown plan result")
}
}

5
pkg/task_logger/task_logger.go
View File

@ -45,6 +45,9 @@ func (s TaskStatus) IsFinished() bool {
return s == TaskStoppedStatus || s == TaskSuccessStatus || s == TaskFailStatus
}
type StatusListener func(status TaskStatus)
type LogListener func(new time.Time, msg string)
type Logger interface {
Log(msg string)
Logf(format string, a ...any)
@ -52,4 +55,6 @@ type Logger interface {
LogfWithTime(now time.Time, format string, a ...any)
LogCmd(cmd *exec.Cmd)
SetStatus(status TaskStatus)
AddStatusListener(l StatusListener)
AddLogListener(l LogListener)
}

18
services/runners/running_job.go
View File

@ -16,6 +16,17 @@ type runningJob struct {
status task_logger.TaskStatus
logRecords []LogRecord
job *tasks.LocalJob
statusListeners []task_logger.StatusListener
logListeners []task_logger.LogListener
}
func (p *runningJob) AddStatusListener(l task_logger.StatusListener) {
p.statusListeners = append(p.statusListeners, l)
}
func (p *runningJob) AddLogListener(l task_logger.LogListener) {
p.logListeners = append(p.logListeners, l)
}
func (p *runningJob) Log(msg string) {
@ -34,6 +45,9 @@ func (p *runningJob) LogWithTime(now time.Time, msg string) {
Message: msg,
},
)
for _, l := range p.logListeners {
l(now, msg)
}
}
func (p *runningJob) LogfWithTime(now time.Time, format string, a ...any) {
@ -55,6 +69,10 @@ func (p *runningJob) SetStatus(status task_logger.TaskStatus) {
p.status = status
p.job.SetStatus(status)
for _, l := range p.statusListeners {
l(status)
}
}
func (p *runningJob) logPipe(reader *bufio.Reader) {

6
services/tasks/LocalJob.go
View File

@ -191,12 +191,6 @@ func (t *LocalJob) getTerraformArgs(username string, incomingVersion *string) (a
args = []string{}
if t.Task.DryRun {
args = append(args, "plan")
} else {
args = append(args, "apply")
}
extraVars, err := t.getEnvironmentExtraVars(username, incomingVersion)
if err != nil {

6
services/tasks/TaskPool.go
View File

@ -34,9 +34,10 @@ type TaskPool struct {
// register channel used to put tasks to queue.
register chan *TaskRunner
// activeProj ???
activeProj map[int]map[int]*TaskRunner
// runningTasks contains tasks with status TaskRunningStatus.
// runningTasks contains tasks with status TaskRunningStatus. Map key is a task ID.
runningTasks map[int]*TaskRunner
// logger channel used to putting log records to database.
@ -193,6 +194,9 @@ func (p *TaskPool) blocks(t *TaskRunner) bool {
}
for _, r := range p.activeProj[t.Task.ProjectID] {
if r.Task.Status.IsFinished() {
continue
}
if r.Template.ID == t.Task.TemplateID {
return true
}

11
services/tasks/TaskRunner.go
View File

@ -38,6 +38,17 @@ type TaskRunner struct {
RunnerID int
Username string
IncomingVersion *string
statusListeners []task_logger.StatusListener
logListeners []task_logger.LogListener
}
func (t *TaskRunner) AddStatusListener(l task_logger.StatusListener) {
t.statusListeners = append(t.statusListeners, l)
}
func (t *TaskRunner) AddLogListener(l task_logger.LogListener) {
t.logListeners = append(t.logListeners, l)
}
func (t *TaskRunner) saveStatus() {

8
services/tasks/TaskRunner_logging.go
View File

@ -40,6 +40,10 @@ func (t *TaskRunner) LogWithTime(now time.Time, msg string) {
output: msg,
time: now,
}
for _, l := range t.logListeners {
l(now, msg)
}
}
func (t *TaskRunner) LogfWithTime(now time.Time, format string, a ...any) {
@ -102,6 +106,10 @@ func (t *TaskRunner) SetStatus(status task_logger.TaskStatus) {
t.sendRocketChatAlert()
t.sendMicrosoftTeamsAlert()
}
for _, l := range t.statusListeners {
l(status)
}
}
func (t *TaskRunner) panicOnError(err error, msg string) {