*: implement run stop

This commit is contained in:
Simone Gotti 2019-03-08 10:02:37 +01:00
parent 6f38c48066
commit f09602cdc3
5 changed files with 67 additions and 3 deletions

View File

@ -336,6 +336,7 @@ type RunActionType string
const ( const (
RunActionTypeRestart RunActionType = "restart" RunActionTypeRestart RunActionType = "restart"
RunActionTypeStop RunActionType = "stop"
) )
type RunActionsRequest struct { type RunActionsRequest struct {
@ -382,6 +383,21 @@ func (h *RunActionsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)
return return
} }
case RunActionTypeStop:
req := &rsapi.RunActionsRequest{
ActionType: rsapi.RunActionTypeStop,
}
resp, err := h.runserviceClient.RunActions(ctx, runID, req)
if err != nil {
if resp != nil && resp.StatusCode == http.StatusNotFound {
http.Error(w, err.Error(), http.StatusNotFound)
return
}
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
} }
} }

View File

@ -484,6 +484,7 @@ type RunActionType string
const ( const (
RunActionTypeChangePhase RunActionType = "changephase" RunActionTypeChangePhase RunActionType = "changephase"
RunActionTypeStop RunActionType = "stop"
) )
type RunActionsRequest struct { type RunActionsRequest struct {
@ -511,7 +512,6 @@ func (h *RunActionsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
runID := vars["runid"] runID := vars["runid"]
// TODO(sgotti) Check authorized call from client
var req RunActionsRequest var req RunActionsRequest
d := json.NewDecoder(r.Body) d := json.NewDecoder(r.Body)
if err := d.Decode(&req); err != nil { if err := d.Decode(&req); err != nil {
@ -530,6 +530,15 @@ func (h *RunActionsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, err.Error(), http.StatusBadRequest) http.Error(w, err.Error(), http.StatusBadRequest)
return return
} }
case RunActionTypeStop:
creq := &command.RunStopRequest{
RunID: runID,
ChangeGroupsUpdateToken: req.ChangeGroupsUpdateToken,
}
if err := h.ch.StopRun(ctx, creq); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
default: default:
http.Error(w, "", http.StatusBadRequest) http.Error(w, "", http.StatusBadRequest)
return return

View File

@ -72,6 +72,39 @@ func (s *CommandHandler) ChangeRunPhase(ctx context.Context, req *RunChangePhase
return errors.Errorf("run %s is not queued but in %q phase", r.ID, r.Phase) return errors.Errorf("run %s is not queued but in %q phase", r.ID, r.Phase)
} }
r.ChangePhase(types.RunPhaseRunning) r.ChangePhase(types.RunPhaseRunning)
case types.RunPhaseFinished:
if r.Phase != types.RunPhaseRunning {
return errors.Errorf("run %s is not running but in %q phase", r.ID, r.Phase)
}
r.Stop = true
}
_, err = store.AtomicPutRun(ctx, s.e, r, "", cgt)
return err
}
type RunStopRequest struct {
RunID string
ChangeGroupsUpdateToken string
}
func (s *CommandHandler) StopRun(ctx context.Context, req *RunStopRequest) error {
cgt, err := types.UnmarshalChangeGroupsUpdateToken(req.ChangeGroupsUpdateToken)
if err != nil {
return err
}
r, _, err := store.GetRun(ctx, s.e, req.RunID)
if err != nil {
return err
}
if r.Phase != types.RunPhaseRunning {
return errors.Errorf("run %s is not running but in %q phase", r.ID, r.Phase)
}
if !r.Result.IsSet() {
// stop only if the result is not setted yet
r.Stop = true
} }
_, err = store.AtomicPutRun(ctx, s.e, r, "", cgt) _, err = store.AtomicPutRun(ctx, s.e, r, "", cgt)
@ -191,6 +224,7 @@ func (s *CommandHandler) recreateRun(ctx context.Context, req *RunCreateRequest)
run.Phase = types.RunPhaseQueued run.Phase = types.RunPhaseQueued
run.Result = types.RunResultUnknown run.Result = types.RunResultUnknown
run.Archived = false run.Archived = false
run.Stop = false
// TODO(sgotti) handle reset tasks // TODO(sgotti) handle reset tasks
// currently we only restart a run resetting al failed tasks // currently we only restart a run resetting al failed tasks

View File

@ -352,6 +352,10 @@ func (s *Scheduler) advanceRun(ctx context.Context, runID string) error {
return nil return nil
} }
if r.Stop {
r.Result = types.RunResultStopped
}
if _, err := store.AtomicPutRun(ctx, s.e, r, "", nil); err != nil { if _, err := store.AtomicPutRun(ctx, s.e, r, "", nil); err != nil {
return err return err
} }

View File

@ -42,8 +42,6 @@ const (
RunPhaseCancelled RunPhase = "cancelled" RunPhaseCancelled RunPhase = "cancelled"
RunPhaseRunning RunPhase = "running" RunPhaseRunning RunPhase = "running"
RunPhaseFinished RunPhase = "finished" RunPhaseFinished RunPhase = "finished"
//RunPhaseSuccess RunPhase = "success"
//RunPhaseFailed RunPhase = "failed"
) )
type RunResult string type RunResult string
@ -99,6 +97,9 @@ type Run struct {
// Result of a Run. // Result of a Run.
Result RunResult `json:"result,omitempty"` Result RunResult `json:"result,omitempty"`
// Stop is used to signal from the scheduler when the run must be stopped
Stop bool `json:"stop,omitempty"`
RunTasks map[string]*RunTask `json:"run_tasks,omitempty"` RunTasks map[string]*RunTask `json:"run_tasks,omitempty"`
EnqueueTime *time.Time `json:"enqueue_time,omitempty"` EnqueueTime *time.Time `json:"enqueue_time,omitempty"`
StartTime *time.Time `json:"start_time,omitempty"` StartTime *time.Time `json:"start_time,omitempty"`