func advanceRun(ctx context.Context, r *types.Run, rc *types.RunConfig, scheduledExecutorTasks []*types.ExecutorTask) error {
	log.Debugf("run: %s", util.Dump(r))
	hasScheduledTasks := len(scheduledExecutorTasks) > 0
	if !r.Result.IsSet() && r.Phase == types.RunPhaseRunning {
		for _, rt := range r.Tasks {
			rct, ok := rc.Tasks[rt.ID]
			log.Debugf("rct: %s", util.Dump(rct))
			if !ok {
				return errors.Errorf("no such run config task with id %s for run config %s", rt.ID, rc.ID)
			}
			if rt.Status == types.RunTaskStatusFailed {
				if !rct.IgnoreFailure {
					log.Debugf("marking run %q as failed is task %q is failed", r.ID, rt.ID)
					r.Result = types.RunResultFailed
					break
				}
			}
		}
	}
	if !r.Result.IsSet() && r.Phase == types.RunPhaseRunning {
		finished := true
		for _, rt := range r.Tasks {
			if !rt.Status.IsFinished() {
				finished = false
			}
		}
		if finished {
			r.Result = types.RunResultSuccess
			return nil
		}
	}
	if !r.Result.IsSet() && r.Phase == types.RunPhaseRunning {
		if r.Stop {
			r.Result = types.RunResultStopped
		}
	}
	if r.Result.IsSet() {
		finished := true
		for _, rt := range r.Tasks {
			if !rt.Status.IsFinished() {
				finished = false
			}
		}
		if finished && !r.Phase.IsFinished() {
			if !hasScheduledTasks {
				r.ChangePhase(types.RunPhaseFinished)
			}
		}
		if r.Phase.IsFinished() {
			for _, rt := range r.Tasks {
				log.Debugf("rt: %s", util.Dump(rt))
				if rt.Status == types.RunTaskStatusNotStarted {
					rt.SetupStep.LogPhase = types.RunTaskFetchPhaseFinished
					for _, s := range rt.Steps {
						s.LogPhase = types.RunTaskFetchPhaseFinished
					}
					for i := range rt.WorkspaceArchivesPhase {
						rt.WorkspaceArchivesPhase[i] = types.RunTaskFetchPhaseFinished
					}
				}
			}
		}
	}
	return nil
}
advanceRun updates the run result and phase. It must be the unique function that
should update them.
package runservice
import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"time"
	"agola.io/agola/internal/datamanager"
	"agola.io/agola/internal/etcd"
	"agola.io/agola/internal/objectstorage"
	"agola.io/agola/internal/runconfig"
	"agola.io/agola/internal/services/runservice/common"
	"agola.io/agola/internal/services/runservice/store"
	"agola.io/agola/internal/util"
	"agola.io/agola/services/runservice/types"
	etcdclientv3 "go.etcd.io/etcd/clientv3"
	"go.etcd.io/etcd/clientv3/concurrency"
	errors "golang.org/x/xerrors"
)
const (
	cacheCleanerInterval     = 1 * 24 * time.Hour
	workspaceCleanerInterval = 1 * 24 * time.Hour
	defaultExecutorNotAliveInterval = 60 * time.Second
)
func taskMatchesParentDependCondition(ctx context.Context, rt *types.RunTask, r *types.Run, rc *types.RunConfig) bool {
	rct := rc.Tasks[rt.ID]
	parents := runconfig.GetParents(rc.Tasks, rct)
	matchedNum := 0
	for _, p := range parents {
		matched := false
		rp := r.Tasks[p.ID]
		conds := runconfig.GetParentDependConditions(rct, p)
		for _, cond := range conds {
			switch cond {
			case types.RunConfigTaskDependConditionOnSuccess:
				if rp.Status == types.RunTaskStatusSuccess {
					matched = true
				}
			case types.RunConfigTaskDependConditionOnFailure:
				if rp.Status == types.RunTaskStatusFailed {
					matched = true
				}
			case types.RunConfigTaskDependConditionOnSkipped:
				if rp.Status == types.RunTaskStatusSkipped {
					matched = true
				}
			}
		}
		if matched {
			matchedNum++
		}
	}
	return len(parents) == matchedNum
}
func advanceRunTasks(ctx context.Context, curRun *types.Run, rc *types.RunConfig, scheduledExecutorTasks []*types.ExecutorTask) (*types.Run, error) {
	log.Debugf("run: %s", util.Dump(curRun))
	log.Debugf("rc: %s", util.Dump(rc))
	newRun := curRun.DeepCopy()
	if newRun.Stop {
		for _, rt := range newRun.Tasks {
			isScheduled := false
			for _, et := range scheduledExecutorTasks {
				if rt.ID == et.ID {
					isScheduled = true
				}
			}
			if isScheduled {
				continue
			}
			if rt.Status == types.RunTaskStatusNotStarted {
				rt.Status = types.RunTaskStatusSkipped
			}
		}
	}
	for _, rt := range newRun.Tasks {
		if rt.Skip {
			continue
		}
		if rt.Status != types.RunTaskStatusNotStarted {
			continue
		}
		rct := rc.Tasks[rt.ID]
		parents := runconfig.GetParents(rc.Tasks, rct)
		if len(parents) > 0 {
			continue
		}
		if curRun.Result.IsSet() {
			isScheduled := false
			for _, et := range scheduledExecutorTasks {
				if rt.ID == et.ID {
					isScheduled = true
				}
			}
			if isScheduled {
				continue
			}
			if rt.Status == types.RunTaskStatusNotStarted {
				rt.Status = types.RunTaskStatusCancelled
			}
		}
	}
	for _, rt := range newRun.Tasks {
		if rt.Skip {
			continue
		}
		if rt.Status != types.RunTaskStatusNotStarted {
			continue
		}
		rct := rc.Tasks[rt.ID]
		parents := runconfig.GetParents(rc.Tasks, rct)
		finishedParents := 0
		for _, p := range parents {
			rp := curRun.Tasks[p.ID]
			if rp.Status.IsFinished() && rp.ArchivesFetchFinished() {
				finishedParents++
			}
		}
		allParentsFinished := finishedParents == len(parents)
		if allParentsFinished {
			matched := taskMatchesParentDependCondition(ctx, rt, curRun, rc)
			skip := !matched
			if skip {
				rt.Status = types.RunTaskStatusSkipped
				continue
			}
			if rct.NeedsApproval && !rt.WaitingApproval && !rt.Approved {
				rt.WaitingApproval = true
			}
		}
	}
	return newRun, nil
}
func getTasksToRun(ctx context.Context, r *types.Run, rc *types.RunConfig) ([]*types.RunTask, error) {
	log.Debugf("run: %s", util.Dump(r))
	log.Debugf("rc: %s", util.Dump(rc))
	tasksToRun := []*types.RunTask{}
	for _, rt := range r.Tasks {
		if rt.Skip {
			continue
		}
		if rt.Status != types.RunTaskStatusNotStarted {
			continue
		}
		rct := rc.Tasks[rt.ID]
		parents := runconfig.GetParents(rc.Tasks, rct)
		finishedParents := 0
		for _, p := range parents {
			rp := r.Tasks[p.ID]
			if rp.Status.IsFinished() && rp.ArchivesFetchFinished() {
				finishedParents++
			}
		}
		allParentsFinished := finishedParents == len(parents)
		if allParentsFinished {
			if !taskMatchesParentDependCondition(ctx, rt, r, rc) {
				continue
			}
			if !rct.NeedsApproval || (rct.NeedsApproval && rt.Approved) {
				tasksToRun = append(tasksToRun, rt)
			}
		}
	}
	return tasksToRun, nil
}
func (s *Runservice) submitRunTasks(ctx context.Context, r *types.Run, rc *types.RunConfig, tasks []*types.RunTask) error {
	log.Debugf("tasksToRun: %s", util.Dump(tasks))
	for _, rt := range tasks {
		rct := rc.Tasks[rt.ID]
		executor, err := s.chooseExecutor(ctx, rct)
		if err != nil {
			return err
		}
		if executor == nil {
			log.Warnf("cannot choose an executor")
			return nil
		}
		et := common.GenExecutorTask(r, rt, rc, executor)
		log.Debugf("et: %s", util.Dump(et))
		tet, err := store.GetExecutorTask(ctx, s.e, et.ID)
		if err != nil && err != etcd.ErrKeyNotFound {
			return err
		}
		if tet != nil {
			continue
		}
		if _, err := store.AtomicPutExecutorTask(ctx, s.e, et); err != nil {
			return err
		}
		if err := s.sendExecutorTask(ctx, et); err != nil {
			return err
		}
	}
	return nil
}
func (s *Runservice) chooseExecutor(ctx context.Context, rct *types.RunConfigTask) (*types.Executor, error) {
	executors, err := store.GetExecutors(ctx, s.e)
	if err != nil {
		return nil, err
	}
	executorTasksCount, err := store.GetExecutorTasksCountByExecutor(ctx, s.e)
	if err != nil {
		return nil, err
	}
	return chooseExecutor(executors, executorTasksCount, rct), nil
}
func chooseExecutor(executors []*types.Executor, executorTasksCount map[string]int, rct *types.RunConfigTask) *types.Executor {
	requiresPrivilegedContainers := false
	for _, c := range rct.Runtime.Containers {
		if c.Privileged {
			requiresPrivilegedContainers = true
			break
		}
	}
	for _, e := range executors {
		if e.LastStatusUpdateTime.Add(defaultExecutorNotAliveInterval).Before(time.Now()) {
			continue
		}
		if requiresPrivilegedContainers && !e.AllowPrivilegedContainers {
			continue
		}
		if rct.Runtime.Arch != "" {
			hasArch := false
			for _, arch := range e.Archs {
				if arch == rct.Runtime.Arch {
					hasArch = true
				}
			}
			if !hasArch {
				continue
			}
		}
		if e.ActiveTasksLimit != 0 {
			activeTasks := executorTasksCount[e.ID]
			if e.ActiveTasks > activeTasks {
				activeTasks = e.ActiveTasks
			}
			if activeTasks >= e.ActiveTasksLimit {
				continue
			}
		}
		return e
	}
	return nil
}
func (s *Runservice) sendExecutorTask(ctx context.Context, et *types.ExecutorTask) error {
	executor, err := store.GetExecutor(ctx, s.e, et.Spec.ExecutorID)
	if err != nil && err != etcd.ErrKeyNotFound {
		return err
	}
	if executor == nil {
		log.Warnf("executor with id %q doesn't exist", et.Spec.ExecutorID)
		return nil
	}
	r, _, err := store.GetRun(ctx, s.e, et.Spec.RunID)
	if err != nil {
		return err
	}
	rc, err := store.OSTGetRunConfig(s.dm, r.ID)
	if err != nil {
		return errors.Errorf("cannot get run config %q: %w", r.ID, err)
	}
	rt, ok := r.Tasks[et.ID]
	if !ok {
		return errors.Errorf("no such run task with id %s for run %s", et.ID, r.ID)
	}
	et = et.DeepCopy()
	et.Spec.ExecutorTaskSpecData = common.GenExecutorTaskSpecData(r, rt, rc)
	etj, err := json.Marshal(et)
	if err != nil {
		return err
	}
	req, err := http.Post(executor.ListenURL+"/api/v1alpha/executor", "", bytes.NewReader(etj))
	if err != nil {
		return err
	}
	if req.StatusCode != http.StatusOK {
		return errors.Errorf("received http status: %d", req.StatusCode)
	}
	return nil
}
func (s *Runservice) compactChangeGroupsLoop(ctx context.Context) {
	for {
		if err := s.compactChangeGroups(ctx); err != nil {
			log.Errorf("err: %+v", err)
		}
		sleepCh := time.NewTimer(1 * time.Second).C
		select {
		case <-ctx.Done():
			return
		case <-sleepCh:
		}
	}
}
func (s *Runservice) compactChangeGroups(ctx context.Context) error {
	session, err := concurrency.NewSession(s.e.Client(), concurrency.WithTTL(5), concurrency.WithContext(ctx))
	if err != nil {
		return err
	}
	defer session.Close()
	m := etcd.NewMutex(session, common.EtcdCompactChangeGroupsLockKey)
	if err := m.TryLock(ctx); err != nil {
		if errors.Is(err, etcd.ErrLocked) {
			return nil
		}
		return err
	}
	defer func() { _ = m.Unlock(ctx) }()
	resp, err := s.e.Client().Get(ctx, common.EtcdChangeGroupMinRevisionKey)
	if err != nil {
		return err
	}
	revision := resp.Kvs[0].ModRevision
	cmp := etcdclientv3.Compare(etcdclientv3.ModRevision(common.EtcdChangeGroupMinRevisionKey), "=", revision)
	then := etcdclientv3.OpPut(common.EtcdChangeGroupMinRevisionKey, "")
	txn := s.e.Client().Txn(ctx).If(cmp).Then(then)
	tresp, err := txn.Commit()
	if err != nil {
		return etcd.FromEtcdError(err)
	}
	if !tresp.Succeeded {
		return errors.Errorf("failed to update change group min revision key due to concurrent update")
	}
	revision = tresp.Header.Revision
	resp, err = s.e.List(ctx, common.EtcdChangeGroupsDir, "", 0)
	if err != nil {
		return err
	}
	for _, kv := range resp.Kvs {
		if kv.ModRevision < revision-common.EtcdChangeGroupMinRevisionRange {
			cmp := etcdclientv3.Compare(etcdclientv3.ModRevision(string(kv.Key)), "=", kv.ModRevision)
			then := etcdclientv3.OpDelete(string(kv.Key))
			txn := s.e.Client().Txn(ctx).If(cmp).Then(then)
			tresp, err := txn.Commit()
			if err != nil {
				return etcd.FromEtcdError(err)
			}
			if !tresp.Succeeded {
				log.Errorf("failed to update change group min revision key due to concurrent update")
			}
		}
	}
	return nil
}
func (s *Runservice) scheduleRun(ctx context.Context, r *types.Run, rc *types.RunConfig) error {
	log.Debugf("r: %s", util.Dump(r))
	prevPhase := r.Phase
	prevResult := r.Result
	scheduledExecutorTasks, err := store.GetExecutorTasksForRun(ctx, s.e, r.ID)
	if err != nil {
		return err
	}
	if err := advanceRun(ctx, r, rc, scheduledExecutorTasks); err != nil {
		return err
	}
	var runEvent *types.RunEvent
	if prevPhase != r.Phase || prevResult != r.Result {
		var err error
		runEvent, err = common.NewRunEvent(ctx, s.e, r.ID, r.Phase, r.Result)
		if err != nil {
			return err
		}
	}
	r, err = store.AtomicPutRun(ctx, s.e, r, runEvent, nil)
	if err != nil {
		return err
	}
	if r.Stop {
		for _, et := range scheduledExecutorTasks {
			et.Spec.Stop = true
			if _, err := store.AtomicPutExecutorTask(ctx, s.e, et); err != nil {
				return err
			}
			if err := s.sendExecutorTask(ctx, et); err != nil {
				return err
			}
		}
	}
	if r.Phase == types.RunPhaseRunning {
		r, err := advanceRunTasks(ctx, r, rc, scheduledExecutorTasks)
		if err != nil {
			return err
		}
		r, err = store.AtomicPutRun(ctx, s.e, r, nil, nil)
		if err != nil {
			return err
		}
		tasksToRun, err := getTasksToRun(ctx, r, rc)
		if err != nil {
			return err
		}
		return s.submitRunTasks(ctx, r, rc, tasksToRun)
	}
	return nil
}