agola/internal/services/runservice/scheduler/api/api.go

626 lines
15 KiB
Go

// Copyright 2019 Sorint.lab
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied
// See the License for the specific language governing permissions and
// limitations under the License.
package api
import (
"bufio"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"strconv"
"github.com/sorintlab/agola/internal/db"
"github.com/sorintlab/agola/internal/etcd"
"github.com/sorintlab/agola/internal/objectstorage"
"github.com/sorintlab/agola/internal/services/runservice/scheduler/command"
"github.com/sorintlab/agola/internal/services/runservice/scheduler/common"
"github.com/sorintlab/agola/internal/services/runservice/scheduler/readdb"
"github.com/sorintlab/agola/internal/services/runservice/scheduler/store"
"github.com/sorintlab/agola/internal/services/runservice/types"
"github.com/sorintlab/agola/internal/wal"
"github.com/gorilla/mux"
"github.com/pkg/errors"
"go.uber.org/zap"
)
type LogsHandler struct {
log *zap.SugaredLogger
e *etcd.Store
lts *objectstorage.ObjStorage
wal *wal.WalManager
}
func NewLogsHandler(logger *zap.Logger, e *etcd.Store, lts *objectstorage.ObjStorage, wal *wal.WalManager) *LogsHandler {
return &LogsHandler{
log: logger.Sugar(),
e: e,
lts: lts,
wal: wal,
}
}
func (h *LogsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
// TODO(sgotti) Check authorized call from client
q := r.URL.Query()
runID := q.Get("runid")
if runID == "" {
http.Error(w, "", http.StatusBadRequest)
return
}
taskID := q.Get("taskid")
if taskID == "" {
http.Error(w, "", http.StatusBadRequest)
return
}
_, setup := q["setup"]
stepStr := q.Get("step")
if !setup && stepStr == "" {
http.Error(w, "", http.StatusBadRequest)
return
}
if setup && stepStr != "" {
http.Error(w, "", http.StatusBadRequest)
return
}
var step int
if stepStr != "" {
var err error
step, err = strconv.Atoi(stepStr)
if err != nil {
http.Error(w, "", http.StatusBadRequest)
return
}
}
follow := false
if _, ok := q["follow"]; ok {
follow = true
}
stream := false
if _, ok := q["stream"]; ok {
stream = true
}
if follow {
stream = true
}
if err, sendError := h.readTaskLogs(ctx, runID, taskID, setup, step, w, follow, stream); err != nil {
h.log.Errorf("err: %+v", err)
if sendError {
switch err.(type) {
case common.ErrNotExist:
http.Error(w, err.Error(), http.StatusNotFound)
default:
http.Error(w, err.Error(), http.StatusInternalServerError)
}
}
}
}
func (h *LogsHandler) readTaskLogs(ctx context.Context, runID, taskID string, setup bool, step int, w http.ResponseWriter, follow, stream bool) (error, bool) {
r, err := store.GetRunEtcdOrLTS(ctx, h.e, h.wal, runID)
if err != nil {
return err, true
}
if r == nil {
return errors.Errorf("no such run with id: %s", runID), true
}
task, ok := r.RunTasks[taskID]
if !ok {
return errors.Errorf("no such task with ID %s in run %s", taskID, runID), true
}
if len(task.Steps) <= step {
return errors.Errorf("no such step for task %s in run %s", taskID, runID), true
}
// if the log has been already fetched use it, otherwise fetch it from the executor
if task.Steps[step].LogPhase == types.RunTaskFetchPhaseFinished {
var logPath string
if setup {
logPath = store.LTSRunTaskSetupLogPath(task.ID)
} else {
logPath = store.LTSRunTaskStepLogPath(task.ID, step)
}
f, err := h.lts.ReadObject(logPath)
if err != nil {
if err == objectstorage.ErrNotExist {
return common.NewErrNotExist(err), true
}
return err, true
}
defer f.Close()
return sendLogs(w, f, stream), false
}
et, err := store.GetExecutorTask(ctx, h.e, task.ID)
if err != nil {
return err, true
}
executor, err := store.GetExecutor(ctx, h.e, et.Status.ExecutorID)
if err != nil && err != etcd.ErrKeyNotFound {
return err, true
}
if executor == nil {
return common.NewErrNotExist(errors.Errorf("executor with id %q doesn't exist", et.Status.ExecutorID)), true
}
var url string
if setup {
url = fmt.Sprintf("%s/api/v1alpha/executor/logs?taskid=%s&setup", executor.ListenURL, taskID)
} else {
url = fmt.Sprintf("%s/api/v1alpha/executor/logs?taskid=%s&step=%d", executor.ListenURL, taskID, step)
}
if follow {
url += "&follow"
}
req, err := http.Get(url)
if err != nil {
return err, true
}
defer req.Body.Close()
if req.StatusCode != http.StatusOK {
if req.StatusCode == http.StatusNotFound {
return common.NewErrNotExist(errors.New("no log on executor")), true
}
return errors.Errorf("received http status: %d", req.StatusCode), true
}
return sendLogs(w, req.Body, stream), false
}
func sendLogs(w http.ResponseWriter, r io.Reader, stream bool) error {
if stream {
w.Header().Set("Content-Type", "text/event-stream")
}
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
br := bufio.NewReader(r)
var flusher http.Flusher
if fl, ok := w.(http.Flusher); ok {
flusher = fl
}
stop := false
for {
if stop {
return nil
}
data, err := br.ReadBytes('\n')
if err != nil {
if err != io.EOF {
return err
}
if len(data) == 0 {
return nil
}
stop = true
}
if stream {
if _, err := w.Write([]byte(fmt.Sprintf("data: %s\n", data))); err != nil {
return err
}
} else {
if _, err := w.Write(data); err != nil {
return err
}
}
if flusher != nil {
flusher.Flush()
}
}
}
type ChangeGroupsUpdateTokensHandler struct {
log *zap.SugaredLogger
readDB *readdb.ReadDB
}
func NewChangeGroupsUpdateTokensHandler(logger *zap.Logger, readDB *readdb.ReadDB) *ChangeGroupsUpdateTokensHandler {
return &ChangeGroupsUpdateTokensHandler{
log: logger.Sugar(),
readDB: readDB,
}
}
func (h *ChangeGroupsUpdateTokensHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
groups := query["changegroup"]
var cgt *types.ChangeGroupsUpdateToken
err := h.readDB.Do(func(tx *db.Tx) error {
var err error
cgt, err = h.readDB.GetChangeGroupsUpdateTokens(tx, groups)
return err
})
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
cgts, err := types.MarshalChangeGroupsUpdateToken(cgt)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
if err := json.NewEncoder(w).Encode(cgts); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
type RunResponse struct {
Run *types.Run `json:"run"`
RunConfig *types.RunConfig `json:"run_config"`
}
type RunHandler struct {
log *zap.SugaredLogger
e *etcd.Store
wal *wal.WalManager
readDB *readdb.ReadDB
}
func NewRunHandler(logger *zap.Logger, e *etcd.Store, wal *wal.WalManager, readDB *readdb.ReadDB) *RunHandler {
return &RunHandler{
log: logger.Sugar(),
e: e,
wal: wal,
readDB: readDB,
}
}
func (h *RunHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
vars := mux.Vars(r)
runID := vars["runid"]
run, _, err := store.GetRun(ctx, h.e, runID)
if err != nil && err != etcd.ErrKeyNotFound {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
if run == nil {
run, err = store.LTSGetRun(h.wal, runID)
if err != nil && err != objectstorage.ErrNotExist {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
if run == nil {
http.Error(w, "", http.StatusNotFound)
return
}
rc, err := store.LTSGetRunConfig(h.wal, run.ID)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
res := &RunResponse{
Run: run,
RunConfig: rc,
}
if err := json.NewEncoder(w).Encode(res); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
const (
DefaultRunsLimit = 25
MaxRunsLimit = 40
)
type GetRunsResponse struct {
Runs []*types.Run `json:"runs"`
ChangeGroupsUpdateToken string `json:"change_groups_update_tokens"`
}
type RunsHandler struct {
log *zap.SugaredLogger
readDB *readdb.ReadDB
}
func NewRunsHandler(logger *zap.Logger, readDB *readdb.ReadDB) *RunsHandler {
return &RunsHandler{
log: logger.Sugar(),
readDB: readDB,
}
}
func (h *RunsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
phaseFilter := types.RunPhaseFromStringSlice(query["phase"])
changeGroups := query["changegroup"]
groups := query["group"]
limitS := query.Get("limit")
limit := DefaultRunsLimit
if limitS != "" {
var err error
limit, err = strconv.Atoi(limitS)
if err != nil {
http.Error(w, "", http.StatusBadRequest)
return
}
}
if limit < 0 {
http.Error(w, "limit must be greater or equal than 0", http.StatusBadRequest)
return
}
if limit > MaxRunsLimit {
limit = MaxRunsLimit
}
sortOrder := types.SortOrderDesc
if _, ok := query["asc"]; ok {
sortOrder = types.SortOrderAsc
}
start := query.Get("start")
var runs []*types.Run
var cgt *types.ChangeGroupsUpdateToken
if len(groups) == 0 {
groups = []string{"."}
}
err := h.readDB.Do(func(tx *db.Tx) error {
if err := h.readDB.PrefetchRuns(tx, groups, phaseFilter, start, limit, sortOrder); err != nil {
h.log.Errorf("err: %+v", err)
return err
}
return nil
})
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
err = h.readDB.Do(func(tx *db.Tx) error {
var err error
runs, err = h.readDB.GetRuns(tx, groups, phaseFilter, start, limit, sortOrder)
if err != nil {
h.log.Errorf("err: %+v", err)
return err
}
cgt, err = h.readDB.GetChangeGroupsUpdateTokens(tx, changeGroups)
return err
})
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
cgts, err := types.MarshalChangeGroupsUpdateToken(cgt)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
resp := &GetRunsResponse{
Runs: runs,
ChangeGroupsUpdateToken: cgts,
}
if err := json.NewEncoder(w).Encode(resp); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
type RunCreateRequest struct {
// new run
RunConfig *types.RunConfig `json:"run_config"`
// existing run
RunID string `json:"run_id"`
RunConfigID string `json:"run_config_id"`
FromStart bool `json:"from_start"`
ResetTasks []string `json:"reset_tasks"`
Group string `json:"group"`
Environment map[string]string `json:"environment"`
Annotations map[string]string `json:"annotations"`
ChangeGroupsUpdateToken string `json:"changeup_update_tokens"`
}
type RunCreateHandler struct {
log *zap.SugaredLogger
ch *command.CommandHandler
}
func NewRunCreateHandler(logger *zap.Logger, ch *command.CommandHandler) *RunCreateHandler {
return &RunCreateHandler{
log: logger.Sugar(),
ch: ch,
}
}
func (h *RunCreateHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
var req RunCreateRequest
d := json.NewDecoder(r.Body)
if err := d.Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
creq := &command.RunCreateRequest{
RunConfig: req.RunConfig,
RunID: req.RunID,
RunConfigID: req.RunConfigID,
FromStart: req.FromStart,
ResetTasks: req.ResetTasks,
Group: req.Group,
Environment: req.Environment,
Annotations: req.Annotations,
ChangeGroupsUpdateToken: req.ChangeGroupsUpdateToken,
}
rb, err := h.ch.CreateRun(ctx, creq)
if err != nil {
h.log.Errorf("err: %+v", err)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
res := &RunResponse{
Run: rb.Run,
RunConfig: rb.Rc,
}
if err := json.NewEncoder(w).Encode(res); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
type RunActionType string
const (
RunActionTypeChangePhase RunActionType = "changephase"
RunActionTypeStop RunActionType = "stop"
)
type RunActionsRequest struct {
ActionType RunActionType `json:"action_type"`
Phase types.RunPhase `json:"phase"`
ChangeGroupsUpdateToken string `json:"change_groups_update_tokens"`
}
type RunActionsHandler struct {
log *zap.SugaredLogger
ch *command.CommandHandler
readDB *readdb.ReadDB
}
func NewRunActionsHandler(logger *zap.Logger, ch *command.CommandHandler) *RunActionsHandler {
return &RunActionsHandler{
log: logger.Sugar(),
ch: ch,
}
}
func (h *RunActionsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
vars := mux.Vars(r)
runID := vars["runid"]
var req RunActionsRequest
d := json.NewDecoder(r.Body)
if err := d.Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
switch req.ActionType {
case RunActionTypeChangePhase:
creq := &command.RunChangePhaseRequest{
RunID: runID,
Phase: req.Phase,
ChangeGroupsUpdateToken: req.ChangeGroupsUpdateToken,
}
if err := h.ch.ChangeRunPhase(ctx, creq); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
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:
http.Error(w, "", http.StatusBadRequest)
return
}
}
type RunTaskActionType string
const (
RunTaskActionTypeApprove RunTaskActionType = "approve"
)
type RunTaskActionsRequest struct {
ActionType RunTaskActionType `json:"action_type"`
ApprovalAnnotations map[string]string `json:"approval_annotations,omitempty"`
ChangeGroupsUpdateToken string `json:"change_groups_update_tokens"`
}
type RunTaskActionsHandler struct {
log *zap.SugaredLogger
ch *command.CommandHandler
readDB *readdb.ReadDB
}
func NewRunTaskActionsHandler(logger *zap.Logger, ch *command.CommandHandler) *RunTaskActionsHandler {
return &RunTaskActionsHandler{
log: logger.Sugar(),
ch: ch,
}
}
func (h *RunTaskActionsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
vars := mux.Vars(r)
runID := vars["runid"]
taskID := vars["taskid"]
// TODO(sgotti) Check authorized call from client
var req RunTaskActionsRequest
d := json.NewDecoder(r.Body)
if err := d.Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
switch req.ActionType {
case RunTaskActionTypeApprove:
creq := &command.RunTaskApproveRequest{
RunID: runID,
TaskID: taskID,
ChangeGroupsUpdateToken: req.ChangeGroupsUpdateToken,
}
if err := h.ch.ApproveRunTask(ctx, creq); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
default:
http.Error(w, "", http.StatusBadRequest)
return
}
}