cc-backend/api/rest.go

440 lines
13 KiB
Go
Raw Normal View History

package api
import (
2022-01-17 13:27:40 +01:00
"bufio"
"context"
2022-02-07 14:56:46 +01:00
"database/sql"
"encoding/json"
2022-02-14 17:31:51 +01:00
"errors"
"fmt"
2022-01-07 09:39:00 +01:00
"io"
"net/http"
2022-01-07 09:39:00 +01:00
"os"
"path/filepath"
2022-02-07 09:57:06 +01:00
"strconv"
"sync"
2022-01-27 09:40:59 +01:00
"github.com/ClusterCockpit/cc-backend/auth"
"github.com/ClusterCockpit/cc-backend/config"
"github.com/ClusterCockpit/cc-backend/graph"
"github.com/ClusterCockpit/cc-backend/graph/model"
2022-01-27 10:35:26 +01:00
"github.com/ClusterCockpit/cc-backend/log"
2022-01-27 09:40:59 +01:00
"github.com/ClusterCockpit/cc-backend/metricdata"
2022-02-06 09:48:31 +01:00
"github.com/ClusterCockpit/cc-backend/repository"
2022-01-27 09:40:59 +01:00
"github.com/ClusterCockpit/cc-backend/schema"
"github.com/gorilla/mux"
)
type RestApi struct {
2022-02-07 07:09:47 +01:00
JobRepository *repository.JobRepository
Resolver *graph.Resolver
MachineStateDir string
OngoingArchivings sync.WaitGroup
}
func (api *RestApi) MountRoutes(r *mux.Router) {
r = r.PathPrefix("/api").Subrouter()
r.StrictSlash(true)
r.HandleFunc("/jobs/start_job/", api.startJob).Methods(http.MethodPost, http.MethodPut)
r.HandleFunc("/jobs/stop_job/", api.stopJob).Methods(http.MethodPost, http.MethodPut)
r.HandleFunc("/jobs/stop_job/{id}", api.stopJob).Methods(http.MethodPost, http.MethodPut)
2022-01-07 09:39:00 +01:00
2022-01-17 13:27:40 +01:00
r.HandleFunc("/jobs/", api.getJobs).Methods(http.MethodGet)
r.HandleFunc("/jobs/{id}", api.getJob).Methods(http.MethodGet)
r.HandleFunc("/jobs/tag_job/{id}", api.tagJob).Methods(http.MethodPost, http.MethodPatch)
2022-01-27 09:29:53 +01:00
r.HandleFunc("/jobs/metrics/{id}", api.getJobMetrics).Methods(http.MethodGet)
2022-01-17 13:27:40 +01:00
if api.MachineStateDir != "" {
r.HandleFunc("/machine_state/{cluster}/{host}", api.getMachineState).Methods(http.MethodGet)
r.HandleFunc("/machine_state/{cluster}/{host}", api.putMachineState).Methods(http.MethodPut, http.MethodPost)
}
}
2022-02-06 09:48:31 +01:00
type StartJobApiResponse struct {
DBID int64 `json:"id"`
}
type StopJobApiRequest struct {
// JobId, ClusterId and StartTime are optional.
// They are only used if no database id was provided.
2022-01-17 13:27:40 +01:00
JobId *int64 `json:"jobId"`
2021-12-20 10:48:58 +01:00
Cluster *string `json:"cluster"`
StartTime *int64 `json:"startTime"`
// Payload
2021-12-17 15:49:22 +01:00
StopTime int64 `json:"stopTime"`
State schema.JobState `json:"jobState"`
}
2022-02-14 17:31:51 +01:00
type ErrorResponse struct {
Status string `json:"status"`
Error string `json:"error"`
}
func handleError(err error, statusCode int, rw http.ResponseWriter) {
log.Printf("REST API error: %s", err.Error())
rw.Header().Add("Content-Type", "application/json")
rw.WriteHeader(statusCode)
json.NewEncoder(rw).Encode(ErrorResponse{
Status: http.StatusText(statusCode),
Error: err.Error(),
})
}
type TagJobApiRequest []*struct {
Name string `json:"name"`
Type string `json:"type"`
}
2022-01-17 13:27:40 +01:00
// Return a list of jobs
func (api *RestApi) getJobs(rw http.ResponseWriter, r *http.Request) {
filter := model.JobFilter{}
for key, vals := range r.URL.Query() {
switch key {
case "state":
for _, s := range vals {
state := schema.JobState(s)
if !state.Valid() {
http.Error(rw, "invalid query parameter value: state", http.StatusBadRequest)
return
}
filter.State = append(filter.State, state)
}
case "cluster":
filter.Cluster = &model.StringInput{Eq: &vals[0]}
default:
http.Error(rw, "invalid query parameter: "+key, http.StatusBadRequest)
return
}
}
results, err := api.Resolver.Query().Jobs(r.Context(), []*model.JobFilter{&filter}, nil, nil)
if err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
return
}
bw := bufio.NewWriter(rw)
defer bw.Flush()
if err := json.NewEncoder(bw).Encode(results.Items); err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
return
}
}
// Return a single job
func (api *RestApi) getJob(rw http.ResponseWriter, r *http.Request) {
id := mux.Vars(r)["id"]
job, err := api.Resolver.Query().Job(r.Context(), id)
if err != nil {
http.Error(rw, err.Error(), http.StatusNotFound)
return
}
job.Tags, err = api.Resolver.Job().Tags(r.Context(), job)
if err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
return
}
rw.Header().Add("Content-Type", "application/json")
rw.WriteHeader(http.StatusOK)
json.NewEncoder(rw).Encode(job)
}
2022-01-17 13:27:40 +01:00
// Add a tag to a job
func (api *RestApi) tagJob(rw http.ResponseWriter, r *http.Request) {
id := mux.Vars(r)["id"]
job, err := api.Resolver.Query().Job(r.Context(), id)
if err != nil {
http.Error(rw, err.Error(), http.StatusNotFound)
return
}
job.Tags, err = api.Resolver.Job().Tags(r.Context(), job)
if err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
return
}
var req TagJobApiRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(rw, err.Error(), http.StatusBadRequest)
return
}
for _, tag := range req {
2022-02-08 12:49:28 +01:00
tagId, err := api.JobRepository.AddTagOrCreate(job.ID, tag.Type, tag.Name)
if err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
return
}
2021-12-17 15:49:22 +01:00
job.Tags = append(job.Tags, &schema.Tag{
ID: tagId,
Type: tag.Type,
Name: tag.Name,
})
}
rw.Header().Add("Content-Type", "application/json")
rw.WriteHeader(http.StatusOK)
json.NewEncoder(rw).Encode(job)
}
2022-01-17 13:27:40 +01:00
// A new job started. The body should be in the `meta.json` format, but some fields required
// there are optional here (e.g. `jobState` defaults to "running").
func (api *RestApi) startJob(rw http.ResponseWriter, r *http.Request) {
if user := auth.GetUser(r.Context()); user != nil && !user.HasRole(auth.RoleApi) {
2022-02-14 17:40:47 +01:00
handleError(fmt.Errorf("missing role: %#v", auth.RoleApi), http.StatusForbidden, rw)
return
}
2021-12-17 15:49:22 +01:00
req := schema.JobMeta{BaseJob: schema.JobDefaults}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
2022-02-14 17:40:47 +01:00
handleError(fmt.Errorf("parsing request body failed: %w", err), http.StatusBadRequest, rw)
return
}
2022-01-24 10:08:47 +01:00
if config.GetClusterConfig(req.Cluster) == nil || config.GetPartition(req.Cluster, req.Partition) == nil {
2022-02-14 17:40:47 +01:00
handleError(fmt.Errorf("cluster or partition does not exist: %#v/%#v", req.Cluster, req.Partition), http.StatusBadRequest, rw)
return
}
2022-01-17 13:27:40 +01:00
// TODO: Do more such checks, be smarter with them.
2021-12-17 15:49:22 +01:00
if len(req.Resources) == 0 || len(req.User) == 0 || req.NumNodes == 0 {
2022-02-14 17:40:47 +01:00
handleError(errors.New("the fields 'resources', 'user' and 'numNodes' are required"), http.StatusBadRequest, rw)
return
}
// Check if combination of (job_id, cluster_id, start_time) already exists:
job, err := api.JobRepository.Find(&req.JobID, &req.Cluster, &req.StartTime)
2022-02-07 14:56:46 +01:00
if err != nil && err != sql.ErrNoRows {
2022-02-14 17:40:47 +01:00
handleError(fmt.Errorf("checking for duplicate failed: %w", err), http.StatusInternalServerError, rw)
return
}
2022-02-07 14:56:46 +01:00
if err != sql.ErrNoRows {
2022-02-14 17:40:47 +01:00
handleError(fmt.Errorf("a job with that jobId, cluster and startTime already exists: dbid: %d", job.ID), http.StatusUnprocessableEntity, rw)
return
}
2022-01-24 10:08:47 +01:00
if req.State == "" {
req.State = schema.JobStateRunning
}
req.RawResources, err = json.Marshal(req.Resources)
2021-12-17 15:49:22 +01:00
if err != nil {
2022-02-14 17:40:47 +01:00
handleError(fmt.Errorf("basically impossible: %w", err), http.StatusBadRequest, rw)
2022-01-27 10:35:26 +01:00
return
2021-12-17 15:49:22 +01:00
}
2022-02-08 12:49:28 +01:00
id, err := api.JobRepository.Start(&req)
if err != nil {
2022-02-14 17:40:47 +01:00
handleError(fmt.Errorf("insert into database failed: %w", err), http.StatusInternalServerError, rw)
return
}
2022-02-08 12:49:28 +01:00
for _, tag := range req.Tags {
if _, err := api.JobRepository.AddTagOrCreate(id, tag.Type, tag.Name); err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
2022-02-14 17:40:47 +01:00
handleError(fmt.Errorf("adding tag to new job %d failed: %w", id, err), http.StatusInternalServerError, rw)
2022-02-08 12:49:28 +01:00
return
}
}
2022-01-27 10:35:26 +01:00
log.Printf("new job (id: %d): cluster=%s, jobId=%d, user=%s, startTime=%d", id, req.Cluster, req.JobID, req.User, req.StartTime)
rw.Header().Add("Content-Type", "application/json")
rw.WriteHeader(http.StatusCreated)
2022-02-06 09:48:31 +01:00
json.NewEncoder(rw).Encode(StartJobApiResponse{
DBID: id,
})
}
2022-01-17 13:27:40 +01:00
// A job has stopped and should be archived.
func (api *RestApi) stopJob(rw http.ResponseWriter, r *http.Request) {
if user := auth.GetUser(r.Context()); user != nil && !user.HasRole(auth.RoleApi) {
2022-02-14 17:31:51 +01:00
handleError(fmt.Errorf("missing role: %#v", auth.RoleApi), http.StatusForbidden, rw)
return
}
// Parse request body
req := StopJobApiRequest{}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
2022-02-14 17:31:51 +01:00
handleError(fmt.Errorf("parsing request body failed: %w", err), http.StatusBadRequest, rw)
return
}
// Fetch job (that will be stopped) from db
id, ok := mux.Vars(r)["id"]
2022-02-07 07:09:47 +01:00
var job *schema.Job
var err error
if ok {
2022-02-08 11:10:05 +01:00
id, e := strconv.ParseInt(id, 10, 64)
if e != nil {
2022-02-14 17:31:51 +01:00
handleError(fmt.Errorf("integer expected in path for id: %w", e), http.StatusBadRequest, rw)
2022-02-08 11:10:05 +01:00
return
2022-02-07 09:57:06 +01:00
}
job, err = api.JobRepository.FindById(id)
} else {
if req.JobId == nil {
handleError(errors.New("the field 'jobId' is required"), http.StatusBadRequest, rw)
return
}
job, err = api.JobRepository.Find(req.JobId, req.Cluster, req.StartTime)
2021-12-17 15:49:22 +01:00
}
if err != nil {
2022-02-14 17:31:51 +01:00
handleError(fmt.Errorf("finding job failed: %w", err), http.StatusUnprocessableEntity, rw)
return
}
// Sanity checks
2021-12-17 15:49:22 +01:00
if job == nil || job.StartTime.Unix() >= req.StopTime || job.State != schema.JobStateRunning {
2022-02-14 17:31:51 +01:00
handleError(errors.New("stopTime must be larger than startTime and only running jobs can be stopped"), http.StatusBadRequest, rw)
return
}
2021-12-17 15:49:22 +01:00
if req.State != "" && !req.State.Valid() {
2022-02-14 17:31:51 +01:00
handleError(fmt.Errorf("invalid job state: %#v", req.State), http.StatusBadRequest, rw)
2021-12-17 15:49:22 +01:00
return
} else {
req.State = schema.JobStateCompleted
}
// Mark job as stopped in the database (update state and duration)
job.Duration = int32(req.StopTime - job.StartTime.Unix())
job.State = req.State
if err := api.JobRepository.Stop(job.ID, job.Duration, job.State, job.MonitoringStatus); err != nil {
handleError(fmt.Errorf("marking job as stopped failed: %w", err), http.StatusInternalServerError, rw)
return
}
2022-01-27 10:35:26 +01:00
log.Printf("archiving job... (dbid: %d): cluster=%s, jobId=%d, user=%s, startTime=%s", job.ID, job.Cluster, job.JobID, job.User, job.StartTime)
// Send a response (with status OK). This means that erros that happen from here on forward
// can *NOT* be communicated to the client. If reading from a MetricDataRepository or
// writing to the filesystem fails, the client will not know.
rw.Header().Add("Content-Type", "application/json")
rw.WriteHeader(http.StatusOK)
json.NewEncoder(rw).Encode(job)
// Monitoring is disabled...
if job.MonitoringStatus == schema.MonitoringStatusDisabled {
return
}
2022-02-15 13:31:54 +01:00
// We need to start a new goroutine as this functions needs to return
// for the response to be flushed to the client.
api.OngoingArchivings.Add(1) // So that a shutdown does not interrupt this goroutine.
go func() {
defer api.OngoingArchivings.Done()
// metricdata.ArchiveJob will fetch all the data from a MetricDataRepository and create meta.json/data.json files
jobMeta, err := metricdata.ArchiveJob(job, context.Background())
if err != nil {
log.Errorf("archiving job (dbid: %d) failed: %s", job.ID, err.Error())
api.JobRepository.UpdateMonitoringStatus(job.ID, schema.MonitoringStatusArchivingFailed)
return
}
// Update the jobs database entry one last time:
if err := api.JobRepository.Archive(job.ID, schema.MonitoringStatusArchivingSuccessful, jobMeta.Statistics); err != nil {
log.Errorf("archiving job (dbid: %d) failed: %s", job.ID, err.Error())
return
}
2022-02-15 13:31:54 +01:00
log.Printf("archiving job (dbid: %d) successful", job.ID)
}()
}
2022-01-07 09:39:00 +01:00
2022-01-27 09:29:53 +01:00
func (api *RestApi) getJobMetrics(rw http.ResponseWriter, r *http.Request) {
id := mux.Vars(r)["id"]
metrics := r.URL.Query()["metric"]
var scopes []schema.MetricScope
for _, scope := range r.URL.Query()["scope"] {
var s schema.MetricScope
if err := s.UnmarshalGQL(scope); err != nil {
http.Error(rw, err.Error(), http.StatusBadRequest)
return
}
scopes = append(scopes, s)
}
rw.Header().Add("Content-Type", "application/json")
rw.WriteHeader(http.StatusOK)
type Respone struct {
Data *struct {
JobMetrics []*model.JobMetricWithName `json:"jobMetrics"`
} `json:"data"`
Error *struct {
Message string `json:"message"`
} `json:"error"`
}
data, err := api.Resolver.Query().JobMetrics(r.Context(), id, metrics, scopes)
if err != nil {
2022-01-27 10:35:26 +01:00
json.NewEncoder(rw).Encode(Respone{
2022-01-27 09:29:53 +01:00
Error: &struct {
Message string "json:\"message\""
}{Message: err.Error()},
2022-01-27 10:35:26 +01:00
})
2022-01-27 09:29:53 +01:00
return
}
2022-01-27 10:35:26 +01:00
json.NewEncoder(rw).Encode(Respone{
2022-01-27 09:29:53 +01:00
Data: &struct {
JobMetrics []*model.JobMetricWithName "json:\"jobMetrics\""
}{JobMetrics: data},
2022-01-27 10:35:26 +01:00
})
2022-01-27 09:29:53 +01:00
}
2022-01-07 09:39:00 +01:00
func (api *RestApi) putMachineState(rw http.ResponseWriter, r *http.Request) {
if api.MachineStateDir == "" {
http.Error(rw, "not enabled", http.StatusNotFound)
return
}
vars := mux.Vars(r)
cluster := vars["cluster"]
host := vars["host"]
dir := filepath.Join(api.MachineStateDir, cluster)
if err := os.MkdirAll(dir, 0755); err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
return
}
filename := filepath.Join(dir, fmt.Sprintf("%s.json", host))
f, err := os.Create(filename)
if err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
return
}
defer f.Close()
if _, err := io.Copy(f, r.Body); err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError)
return
}
rw.WriteHeader(http.StatusCreated)
}
func (api *RestApi) getMachineState(rw http.ResponseWriter, r *http.Request) {
if api.MachineStateDir == "" {
http.Error(rw, "not enabled", http.StatusNotFound)
return
}
vars := mux.Vars(r)
filename := filepath.Join(api.MachineStateDir, vars["cluster"], fmt.Sprintf("%s.json", vars["host"]))
// Sets the content-type and 'Last-Modified' Header and so on automatically
http.ServeFile(rw, r, filename)
}