mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2024-12-27 21:59:05 +01:00
1417 lines
50 KiB
Go
1417 lines
50 KiB
Go
// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
|
|
// All rights reserved.
|
|
// Use of this source code is governed by a MIT-style
|
|
// license that can be found in the LICENSE file.
|
|
package api
|
|
|
|
import (
|
|
"bufio"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/ClusterCockpit/cc-backend/internal/archiver"
|
|
"github.com/ClusterCockpit/cc-backend/internal/auth"
|
|
"github.com/ClusterCockpit/cc-backend/internal/config"
|
|
"github.com/ClusterCockpit/cc-backend/internal/graph"
|
|
"github.com/ClusterCockpit/cc-backend/internal/graph/model"
|
|
"github.com/ClusterCockpit/cc-backend/internal/importer"
|
|
"github.com/ClusterCockpit/cc-backend/internal/metricDataDispatcher"
|
|
"github.com/ClusterCockpit/cc-backend/internal/repository"
|
|
"github.com/ClusterCockpit/cc-backend/internal/util"
|
|
"github.com/ClusterCockpit/cc-backend/pkg/archive"
|
|
"github.com/ClusterCockpit/cc-backend/pkg/log"
|
|
"github.com/ClusterCockpit/cc-backend/pkg/schema"
|
|
"github.com/gorilla/mux"
|
|
)
|
|
|
|
// @title ClusterCockpit REST API
|
|
// @version 1.0.0
|
|
// @description API for batch job control.
|
|
|
|
// @contact.name ClusterCockpit Project
|
|
// @contact.url https://github.com/ClusterCockpit
|
|
// @contact.email support@clustercockpit.org
|
|
|
|
// @license.name MIT License
|
|
// @license.url https://opensource.org/licenses/MIT
|
|
|
|
// @host localhost:8080
|
|
// @basePath /api
|
|
|
|
// @securityDefinitions.apikey ApiKeyAuth
|
|
// @in header
|
|
// @name X-Auth-Token
|
|
|
|
type RestApi struct {
|
|
JobRepository *repository.JobRepository
|
|
Authentication *auth.Authentication
|
|
MachineStateDir string
|
|
RepositoryMutex sync.Mutex
|
|
}
|
|
|
|
func New() *RestApi {
|
|
return &RestApi{
|
|
JobRepository: repository.GetJobRepository(),
|
|
MachineStateDir: config.Keys.MachineStateDir,
|
|
Authentication: auth.GetAuthInstance(),
|
|
}
|
|
}
|
|
|
|
func (api *RestApi) MountApiRoutes(r *mux.Router) {
|
|
r.StrictSlash(true)
|
|
|
|
r.HandleFunc("/jobs/start_job/", api.startJob).Methods(http.MethodPost, http.MethodPut)
|
|
r.HandleFunc("/jobs/stop_job/", api.stopJobByRequest).Methods(http.MethodPost, http.MethodPut)
|
|
// r.HandleFunc("/jobs/import/", api.importJob).Methods(http.MethodPost, http.MethodPut)
|
|
|
|
r.HandleFunc("/jobs/", api.getJobs).Methods(http.MethodGet)
|
|
r.HandleFunc("/jobs/{id}", api.getJobById).Methods(http.MethodPost)
|
|
r.HandleFunc("/jobs/{id}", api.getCompleteJobById).Methods(http.MethodGet)
|
|
r.HandleFunc("/jobs/tag_job/{id}", api.tagJob).Methods(http.MethodPost, http.MethodPatch)
|
|
r.HandleFunc("/jobs/edit_meta/{id}", api.editMeta).Methods(http.MethodPost, http.MethodPatch)
|
|
r.HandleFunc("/jobs/metrics/{id}", api.getJobMetrics).Methods(http.MethodGet)
|
|
r.HandleFunc("/jobs/delete_job/", api.deleteJobByRequest).Methods(http.MethodDelete)
|
|
r.HandleFunc("/jobs/delete_job/{id}", api.deleteJobById).Methods(http.MethodDelete)
|
|
r.HandleFunc("/jobs/delete_job_before/{ts}", api.deleteJobBefore).Methods(http.MethodDelete)
|
|
|
|
r.HandleFunc("/clusters/", api.getClusters).Methods(http.MethodGet)
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
func (api *RestApi) MountUserApiRoutes(r *mux.Router) {
|
|
r.StrictSlash(true)
|
|
|
|
r.HandleFunc("/jobs/", api.getJobs).Methods(http.MethodGet)
|
|
r.HandleFunc("/jobs/{id}", api.getJobById).Methods(http.MethodPost)
|
|
r.HandleFunc("/jobs/{id}", api.getCompleteJobById).Methods(http.MethodGet)
|
|
r.HandleFunc("/jobs/metrics/{id}", api.getJobMetrics).Methods(http.MethodGet)
|
|
}
|
|
|
|
func (api *RestApi) MountConfigApiRoutes(r *mux.Router) {
|
|
r.StrictSlash(true)
|
|
|
|
if api.Authentication != nil {
|
|
r.HandleFunc("/roles/", api.getRoles).Methods(http.MethodGet)
|
|
r.HandleFunc("/users/", api.createUser).Methods(http.MethodPost, http.MethodPut)
|
|
r.HandleFunc("/users/", api.getUsers).Methods(http.MethodGet)
|
|
r.HandleFunc("/users/", api.deleteUser).Methods(http.MethodDelete)
|
|
r.HandleFunc("/user/{id}", api.updateUser).Methods(http.MethodPost)
|
|
}
|
|
}
|
|
|
|
func (api *RestApi) MountFrontendApiRoutes(r *mux.Router) {
|
|
r.StrictSlash(true)
|
|
|
|
if api.Authentication != nil {
|
|
r.HandleFunc("/jwt/", api.getJWT).Methods(http.MethodGet)
|
|
r.HandleFunc("/configuration/", api.updateConfiguration).Methods(http.MethodPost)
|
|
}
|
|
}
|
|
|
|
// StartJobApiResponse model
|
|
type StartJobApiResponse struct {
|
|
// Database ID of new job
|
|
DBID int64 `json:"id"`
|
|
}
|
|
|
|
// DeleteJobApiResponse model
|
|
type DeleteJobApiResponse struct {
|
|
Message string `json:"msg"`
|
|
}
|
|
|
|
// UpdateUserApiResponse model
|
|
type UpdateUserApiResponse struct {
|
|
Message string `json:"msg"`
|
|
}
|
|
|
|
// StopJobApiRequest model
|
|
type StopJobApiRequest struct {
|
|
JobId *int64 `json:"jobId" example:"123000"`
|
|
Cluster *string `json:"cluster" example:"fritz"`
|
|
StartTime *int64 `json:"startTime" example:"1649723812"`
|
|
State schema.JobState `json:"jobState" validate:"required" example:"completed"`
|
|
StopTime int64 `json:"stopTime" validate:"required" example:"1649763839"`
|
|
}
|
|
|
|
// DeleteJobApiRequest model
|
|
type DeleteJobApiRequest struct {
|
|
JobId *int64 `json:"jobId" validate:"required" example:"123000"` // Cluster Job ID of job
|
|
Cluster *string `json:"cluster" example:"fritz"` // Cluster of job
|
|
StartTime *int64 `json:"startTime" example:"1649723812"` // Start Time of job as epoch
|
|
}
|
|
|
|
// GetJobsApiResponse model
|
|
type GetJobsApiResponse struct {
|
|
Jobs []*schema.JobMeta `json:"jobs"` // Array of jobs
|
|
Items int `json:"items"` // Number of jobs returned
|
|
Page int `json:"page"` // Page id returned
|
|
}
|
|
|
|
// GetClustersApiResponse model
|
|
type GetClustersApiResponse struct {
|
|
Clusters []*schema.Cluster `json:"clusters"` // Array of clusters
|
|
}
|
|
|
|
// ErrorResponse model
|
|
type ErrorResponse struct {
|
|
// Statustext of Errorcode
|
|
Status string `json:"status"`
|
|
Error string `json:"error"` // Error Message
|
|
}
|
|
|
|
// ApiTag model
|
|
type ApiTag struct {
|
|
// Tag Type
|
|
Type string `json:"type" example:"Debug"`
|
|
Name string `json:"name" example:"Testjob"` // Tag Name
|
|
Scope string `json:"scope" example:"global"` // Tag Scope for Frontend Display
|
|
}
|
|
|
|
// ApiMeta model
|
|
type EditMetaRequest struct {
|
|
Key string `json:"key" example:"jobScript"`
|
|
Value string `json:"value" example:"bash script"`
|
|
}
|
|
|
|
type TagJobApiRequest []*ApiTag
|
|
|
|
type GetJobApiRequest []string
|
|
|
|
type GetJobApiResponse struct {
|
|
Meta *schema.Job
|
|
Data []*JobMetricWithName
|
|
}
|
|
|
|
type GetCompleteJobApiResponse struct {
|
|
Meta *schema.Job
|
|
Data schema.JobData
|
|
}
|
|
|
|
type JobMetricWithName struct {
|
|
Metric *schema.JobMetric `json:"metric"`
|
|
Name string `json:"name"`
|
|
Scope schema.MetricScope `json:"scope"`
|
|
}
|
|
|
|
type ApiReturnedUser struct {
|
|
Username string `json:"username"`
|
|
Name string `json:"name"`
|
|
Roles []string `json:"roles"`
|
|
Email string `json:"email"`
|
|
Projects []string `json:"projects"`
|
|
}
|
|
|
|
func handleError(err error, statusCode int, rw http.ResponseWriter) {
|
|
log.Warnf("REST 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(),
|
|
})
|
|
}
|
|
|
|
func decode(r io.Reader, val interface{}) error {
|
|
dec := json.NewDecoder(r)
|
|
dec.DisallowUnknownFields()
|
|
return dec.Decode(val)
|
|
}
|
|
|
|
func securedCheck(r *http.Request) error {
|
|
user := repository.GetUserFromContext(r.Context())
|
|
if user == nil {
|
|
return fmt.Errorf("no user in context")
|
|
}
|
|
|
|
if user.AuthType == schema.AuthToken {
|
|
// If nothing declared in config: deny all request to this endpoint
|
|
if config.Keys.ApiAllowedIPs == nil || len(config.Keys.ApiAllowedIPs) == 0 {
|
|
return fmt.Errorf("missing configuration key ApiAllowedIPs")
|
|
}
|
|
|
|
if config.Keys.ApiAllowedIPs[0] == "*" {
|
|
return nil
|
|
}
|
|
|
|
// extract IP address
|
|
IPAddress := r.Header.Get("X-Real-Ip")
|
|
if IPAddress == "" {
|
|
IPAddress = r.Header.Get("X-Forwarded-For")
|
|
}
|
|
if IPAddress == "" {
|
|
IPAddress = r.RemoteAddr
|
|
}
|
|
|
|
if strings.Contains(IPAddress, ":") {
|
|
IPAddress = strings.Split(IPAddress, ":")[0]
|
|
}
|
|
|
|
// check if IP is allowed
|
|
if !util.Contains(config.Keys.ApiAllowedIPs, IPAddress) {
|
|
return fmt.Errorf("unknown ip: %v", IPAddress)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// getClusters godoc
|
|
// @summary Lists all cluster configs
|
|
// @tags Cluster query
|
|
// @description Get a list of all cluster configs. Specific cluster can be requested using query parameter.
|
|
// @produce json
|
|
// @param cluster query string false "Job Cluster"
|
|
// @success 200 {object} api.GetClustersApiResponse "Array of clusters"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /clusters/ [get]
|
|
func (api *RestApi) getClusters(rw http.ResponseWriter, r *http.Request) {
|
|
if user := repository.GetUserFromContext(r.Context()); user != nil &&
|
|
!user.HasRole(schema.RoleApi) {
|
|
|
|
handleError(fmt.Errorf("missing role: %v", schema.GetRoleString(schema.RoleApi)), http.StatusForbidden, rw)
|
|
return
|
|
}
|
|
|
|
rw.Header().Add("Content-Type", "application/json")
|
|
bw := bufio.NewWriter(rw)
|
|
defer bw.Flush()
|
|
|
|
var clusters []*schema.Cluster
|
|
|
|
if r.URL.Query().Has("cluster") {
|
|
name := r.URL.Query().Get("cluster")
|
|
cluster := archive.GetCluster(name)
|
|
if cluster == nil {
|
|
handleError(fmt.Errorf("unknown cluster: %s", name), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
clusters = append(clusters, cluster)
|
|
} else {
|
|
clusters = archive.Clusters
|
|
}
|
|
|
|
payload := GetClustersApiResponse{
|
|
Clusters: clusters,
|
|
}
|
|
|
|
if err := json.NewEncoder(bw).Encode(payload); err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
}
|
|
|
|
// getJobs godoc
|
|
// @summary Lists all jobs
|
|
// @tags Job query
|
|
// @description Get a list of all jobs. Filters can be applied using query parameters.
|
|
// @description Number of results can be limited by page. Results are sorted by descending startTime.
|
|
// @produce json
|
|
// @param state query string false "Job State" Enums(running, completed, failed, cancelled, stopped, timeout)
|
|
// @param cluster query string false "Job Cluster"
|
|
// @param start-time query string false "Syntax: '$from-$to', as unix epoch timestamps in seconds"
|
|
// @param items-per-page query int false "Items per page (Default: 25)"
|
|
// @param page query int false "Page Number (Default: 1)"
|
|
// @param with-metadata query bool false "Include metadata (e.g. jobScript) in response"
|
|
// @success 200 {object} api.GetJobsApiResponse "Job array and page info"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/ [get]
|
|
func (api *RestApi) getJobs(rw http.ResponseWriter, r *http.Request) {
|
|
withMetadata := false
|
|
filter := &model.JobFilter{}
|
|
page := &model.PageRequest{ItemsPerPage: 25, Page: 1}
|
|
order := &model.OrderByInput{Field: "startTime", Order: model.SortDirectionEnumDesc}
|
|
|
|
for key, vals := range r.URL.Query() {
|
|
switch key {
|
|
case "state":
|
|
for _, s := range vals {
|
|
state := schema.JobState(s)
|
|
if !state.Valid() {
|
|
handleError(fmt.Errorf("invalid query parameter value: state"),
|
|
http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
filter.State = append(filter.State, state)
|
|
}
|
|
case "cluster":
|
|
filter.Cluster = &model.StringInput{Eq: &vals[0]}
|
|
case "start-time":
|
|
st := strings.Split(vals[0], "-")
|
|
if len(st) != 2 {
|
|
handleError(fmt.Errorf("invalid query parameter value: startTime"),
|
|
http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
from, err := strconv.ParseInt(st[0], 10, 64)
|
|
if err != nil {
|
|
handleError(err, http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
to, err := strconv.ParseInt(st[1], 10, 64)
|
|
if err != nil {
|
|
handleError(err, http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
ufrom, uto := time.Unix(from, 0), time.Unix(to, 0)
|
|
filter.StartTime = &schema.TimeRange{From: &ufrom, To: &uto}
|
|
case "page":
|
|
x, err := strconv.Atoi(vals[0])
|
|
if err != nil {
|
|
handleError(err, http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
page.Page = x
|
|
case "items-per-page":
|
|
x, err := strconv.Atoi(vals[0])
|
|
if err != nil {
|
|
handleError(err, http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
page.ItemsPerPage = x
|
|
case "with-metadata":
|
|
withMetadata = true
|
|
default:
|
|
handleError(fmt.Errorf("invalid query parameter: %s", key),
|
|
http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
}
|
|
|
|
jobs, err := api.JobRepository.QueryJobs(r.Context(), []*model.JobFilter{filter}, page, order)
|
|
if err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
|
|
results := make([]*schema.JobMeta, 0, len(jobs))
|
|
for _, job := range jobs {
|
|
if withMetadata {
|
|
if _, err = api.JobRepository.FetchMetadata(job); err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
}
|
|
|
|
res := &schema.JobMeta{
|
|
ID: &job.ID,
|
|
BaseJob: job.BaseJob,
|
|
StartTime: job.StartTime.Unix(),
|
|
}
|
|
|
|
res.Tags, err = api.JobRepository.GetTags(repository.GetUserFromContext(r.Context()), &job.ID)
|
|
if err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
|
|
if res.MonitoringStatus == schema.MonitoringStatusArchivingSuccessful {
|
|
res.Statistics, err = archive.GetStatistics(job)
|
|
if err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
}
|
|
|
|
results = append(results, res)
|
|
}
|
|
|
|
log.Debugf("/api/jobs: %d jobs returned", len(results))
|
|
rw.Header().Add("Content-Type", "application/json")
|
|
bw := bufio.NewWriter(rw)
|
|
defer bw.Flush()
|
|
|
|
payload := GetJobsApiResponse{
|
|
Jobs: results,
|
|
Items: page.ItemsPerPage,
|
|
Page: page.Page,
|
|
}
|
|
|
|
if err := json.NewEncoder(bw).Encode(payload); err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
}
|
|
|
|
// getCompleteJobById godoc
|
|
// @summary Get job meta and optional all metric data
|
|
// @tags Job query
|
|
// @description Job to get is specified by database ID
|
|
// @description Returns full job resource information according to 'JobMeta' scheme and all metrics according to 'JobData'.
|
|
// @produce json
|
|
// @param id path int true "Database ID of Job"
|
|
// @param all-metrics query bool false "Include all available metrics"
|
|
// @success 200 {object} api.GetJobApiResponse "Job resource"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 404 {object} api.ErrorResponse "Resource not found"
|
|
// @failure 422 {object} api.ErrorResponse "Unprocessable Entity: finding job failed: sql: no rows in result set"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/{id} [get]
|
|
func (api *RestApi) getCompleteJobById(rw http.ResponseWriter, r *http.Request) {
|
|
// Fetch job from db
|
|
id, ok := mux.Vars(r)["id"]
|
|
var job *schema.Job
|
|
var err error
|
|
if ok {
|
|
id, e := strconv.ParseInt(id, 10, 64)
|
|
if e != nil {
|
|
handleError(fmt.Errorf("integer expected in path for id: %w", e), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
job, err = api.JobRepository.FindById(r.Context(), id) // Get Job from Repo by ID
|
|
} else {
|
|
handleError(fmt.Errorf("the parameter 'id' is required"), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
if err != nil {
|
|
handleError(fmt.Errorf("finding job with db id %s failed: %w", id, err), http.StatusUnprocessableEntity, rw)
|
|
return
|
|
}
|
|
|
|
job.Tags, err = api.JobRepository.GetTags(repository.GetUserFromContext(r.Context()), &job.ID)
|
|
if err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
|
|
}
|
|
if _, err = api.JobRepository.FetchMetadata(job); err != nil {
|
|
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
|
|
var scopes []schema.MetricScope
|
|
|
|
if job.NumNodes == 1 {
|
|
scopes = []schema.MetricScope{"core"}
|
|
} else {
|
|
scopes = []schema.MetricScope{"node"}
|
|
}
|
|
|
|
var data schema.JobData
|
|
|
|
metricConfigs := archive.GetCluster(job.Cluster).MetricConfig
|
|
resolution := 0
|
|
|
|
for _, mc := range metricConfigs {
|
|
resolution = max(resolution, mc.Timestep)
|
|
}
|
|
|
|
if r.URL.Query().Get("all-metrics") == "true" {
|
|
data, err = metricDataDispatcher.LoadData(job, nil, scopes, r.Context(), resolution)
|
|
if err != nil {
|
|
log.Warnf("REST: error while loading all-metrics job data for JobID %d on %s", job.JobID, job.Cluster)
|
|
return
|
|
}
|
|
}
|
|
|
|
log.Debugf("/api/job/%s: get job %d", id, job.JobID)
|
|
rw.Header().Add("Content-Type", "application/json")
|
|
bw := bufio.NewWriter(rw)
|
|
defer bw.Flush()
|
|
|
|
payload := GetCompleteJobApiResponse{
|
|
Meta: job,
|
|
Data: data,
|
|
}
|
|
|
|
if err := json.NewEncoder(bw).Encode(payload); err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
}
|
|
|
|
// getJobById godoc
|
|
// @summary Get job meta and configurable metric data
|
|
// @tags Job query
|
|
// @description Job to get is specified by database ID
|
|
// @description Returns full job resource information according to 'JobMeta' scheme and all metrics according to 'JobData'.
|
|
// @accept json
|
|
// @produce json
|
|
// @param id path int true "Database ID of Job"
|
|
// @param request body api.GetJobApiRequest true "Array of metric names"
|
|
// @success 200 {object} api.GetJobApiResponse "Job resource"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 404 {object} api.ErrorResponse "Resource not found"
|
|
// @failure 422 {object} api.ErrorResponse "Unprocessable Entity: finding job failed: sql: no rows in result set"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/{id} [post]
|
|
func (api *RestApi) getJobById(rw http.ResponseWriter, r *http.Request) {
|
|
// Fetch job from db
|
|
id, ok := mux.Vars(r)["id"]
|
|
var job *schema.Job
|
|
var err error
|
|
if ok {
|
|
id, e := strconv.ParseInt(id, 10, 64)
|
|
if e != nil {
|
|
handleError(fmt.Errorf("integer expected in path for id: %w", e), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
job, err = api.JobRepository.FindById(r.Context(), id)
|
|
} else {
|
|
handleError(errors.New("the parameter 'id' is required"), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
if err != nil {
|
|
handleError(fmt.Errorf("finding job with db id %s failed: %w", id, err), http.StatusUnprocessableEntity, rw)
|
|
return
|
|
}
|
|
|
|
job.Tags, err = api.JobRepository.GetTags(repository.GetUserFromContext(r.Context()), &job.ID)
|
|
if err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
|
|
}
|
|
if _, err = api.JobRepository.FetchMetadata(job); err != nil {
|
|
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
|
|
var metrics GetJobApiRequest
|
|
if err = decode(r.Body, &metrics); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
var scopes []schema.MetricScope
|
|
|
|
if job.NumNodes == 1 {
|
|
scopes = []schema.MetricScope{"core"}
|
|
} else {
|
|
scopes = []schema.MetricScope{"node"}
|
|
}
|
|
|
|
metricConfigs := archive.GetCluster(job.Cluster).MetricConfig
|
|
resolution := 0
|
|
|
|
for _, mc := range metricConfigs {
|
|
resolution = max(resolution, mc.Timestep)
|
|
}
|
|
|
|
data, err := metricDataDispatcher.LoadData(job, metrics, scopes, r.Context(), resolution)
|
|
if err != nil {
|
|
log.Warnf("REST: error while loading job data for JobID %d on %s", job.JobID, job.Cluster)
|
|
return
|
|
}
|
|
|
|
res := []*JobMetricWithName{}
|
|
for name, md := range data {
|
|
for scope, metric := range md {
|
|
res = append(res, &JobMetricWithName{
|
|
Name: name,
|
|
Scope: scope,
|
|
Metric: metric,
|
|
})
|
|
}
|
|
}
|
|
|
|
log.Debugf("/api/job/%s: get job %d", id, job.JobID)
|
|
rw.Header().Add("Content-Type", "application/json")
|
|
bw := bufio.NewWriter(rw)
|
|
defer bw.Flush()
|
|
|
|
payload := GetJobApiResponse{
|
|
Meta: job,
|
|
Data: res,
|
|
}
|
|
|
|
if err := json.NewEncoder(bw).Encode(payload); err != nil {
|
|
handleError(err, http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
}
|
|
|
|
// editMeta godoc
|
|
// @summary Edit meta-data json
|
|
// @tags Job add and modify
|
|
// @description Edit key value pairs in job metadata json
|
|
// @description If a key already exists its content will be overwritten
|
|
// @accept json
|
|
// @produce json
|
|
// @param id path int true "Job Database ID"
|
|
// @param request body api.EditMetaRequest true "Kay value pair to add"
|
|
// @success 200 {object} schema.Job "Updated job resource"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 404 {object} api.ErrorResponse "Job does not exist"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/edit_meta/{id} [post]
|
|
func (api *RestApi) editMeta(rw http.ResponseWriter, r *http.Request) {
|
|
id, err := strconv.ParseInt(mux.Vars(r)["id"], 10, 64)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
job, err := api.JobRepository.FindById(r.Context(), id)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusNotFound)
|
|
return
|
|
}
|
|
|
|
var req EditMetaRequest
|
|
if err := decode(r.Body, &req); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if err := api.JobRepository.UpdateMetadata(job, req.Key, req.Value); 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)
|
|
}
|
|
|
|
// tagJob godoc
|
|
// @summary Adds one or more tags to a job
|
|
// @tags Job add and modify
|
|
// @description Adds tag(s) to a job specified by DB ID. Name and Type of Tag(s) can be chosen freely.
|
|
// @description Tag Scope for frontend visibility will default to "global" if none entered, other options: "admin" or specific username.
|
|
// @description If tagged job is already finished: Tag will be written directly to respective archive files.
|
|
// @accept json
|
|
// @produce json
|
|
// @param id path int true "Job Database ID"
|
|
// @param request body api.TagJobApiRequest true "Array of tag-objects to add"
|
|
// @success 200 {object} schema.Job "Updated job resource"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 404 {object} api.ErrorResponse "Job or tag does not exist"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/tag_job/{id} [post]
|
|
func (api *RestApi) tagJob(rw http.ResponseWriter, r *http.Request) {
|
|
id, err := strconv.ParseInt(mux.Vars(r)["id"], 10, 64)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
job, err := api.JobRepository.FindById(r.Context(), id)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusNotFound)
|
|
return
|
|
}
|
|
|
|
job.Tags, err = api.JobRepository.GetTags(repository.GetUserFromContext(r.Context()), &job.ID)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
var req TagJobApiRequest
|
|
if err := decode(r.Body, &req); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
for _, tag := range req {
|
|
tagId, err := api.JobRepository.AddTagOrCreate(repository.GetUserFromContext(r.Context()), job.ID, tag.Type, tag.Name, tag.Scope)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
job.Tags = append(job.Tags, &schema.Tag{
|
|
ID: tagId,
|
|
Type: tag.Type,
|
|
Name: tag.Name,
|
|
Scope: tag.Scope,
|
|
})
|
|
}
|
|
|
|
rw.Header().Add("Content-Type", "application/json")
|
|
rw.WriteHeader(http.StatusOK)
|
|
json.NewEncoder(rw).Encode(job)
|
|
}
|
|
|
|
// startJob godoc
|
|
// @summary Adds a new job as "running"
|
|
// @tags Job add and modify
|
|
// @description Job specified in request body will be saved to database as "running" with new DB ID.
|
|
// @description Job specifications follow the 'JobMeta' scheme, API will fail to execute if requirements are not met.
|
|
// @accept json
|
|
// @produce json
|
|
// @param request body schema.JobMeta true "Job to add"
|
|
// @success 201 {object} api.StartJobApiResponse "Job added successfully"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 422 {object} api.ErrorResponse "Unprocessable Entity: The combination of jobId, clusterId and startTime does already exist"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/start_job/ [post]
|
|
func (api *RestApi) startJob(rw http.ResponseWriter, r *http.Request) {
|
|
req := schema.JobMeta{BaseJob: schema.JobDefaults}
|
|
if err := decode(r.Body, &req); err != nil {
|
|
handleError(fmt.Errorf("parsing request body failed: %w", err), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
if req.State == "" {
|
|
req.State = schema.JobStateRunning
|
|
}
|
|
if err := importer.SanityChecks(&req.BaseJob); err != nil {
|
|
handleError(err, http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
// Check if combination of (job_id, cluster_id, start_time) already exists:
|
|
jobs, err := api.JobRepository.FindAll(&req.JobID, &req.Cluster, nil)
|
|
if err != nil && err != sql.ErrNoRows {
|
|
handleError(fmt.Errorf("checking for duplicate failed: %w", err), http.StatusInternalServerError, rw)
|
|
return
|
|
} else if err == nil {
|
|
for _, job := range jobs {
|
|
if (req.StartTime - job.StartTimeUnix) < 86400 {
|
|
handleError(fmt.Errorf("a job with that jobId, cluster and startTime already exists: dbid: %d, jobid: %d", job.ID, job.JobID), http.StatusUnprocessableEntity, rw)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
repository.TriggerJobStart(repository.JobWithUser{Job: &req, User: repository.GetUserFromContext(r.Context())})
|
|
|
|
id, err := api.JobRepository.Start(&req)
|
|
if err != nil {
|
|
handleError(fmt.Errorf("insert into database failed: %w", err), http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
|
|
for _, tag := range req.Tags {
|
|
if _, err := api.JobRepository.AddTagOrCreate(repository.GetUserFromContext(r.Context()), id, tag.Type, tag.Name, tag.Scope); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusInternalServerError)
|
|
handleError(fmt.Errorf("adding tag to new job %d failed: %w", id, err), http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
}
|
|
|
|
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)
|
|
json.NewEncoder(rw).Encode(StartJobApiResponse{
|
|
DBID: id,
|
|
})
|
|
}
|
|
|
|
// stopJobByRequest godoc
|
|
// @summary Marks job as completed and triggers archiving
|
|
// @tags Job add and modify
|
|
// @description Job to stop is specified by request body. All fields are required in this case.
|
|
// @description Returns full job resource information according to 'JobMeta' scheme.
|
|
// @produce json
|
|
// @param request body api.StopJobApiRequest true "All fields required"
|
|
// @success 200 {object} schema.JobMeta "Success message"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 404 {object} api.ErrorResponse "Resource not found"
|
|
// @failure 422 {object} api.ErrorResponse "Unprocessable Entity: finding job failed: sql: no rows in result set"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/stop_job/ [post]
|
|
func (api *RestApi) stopJobByRequest(rw http.ResponseWriter, r *http.Request) {
|
|
// Parse request body
|
|
req := StopJobApiRequest{}
|
|
if err := decode(r.Body, &req); err != nil {
|
|
handleError(fmt.Errorf("parsing request body failed: %w", err), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
// Fetch job (that will be stopped) from db
|
|
var job *schema.Job
|
|
var err error
|
|
if req.JobId == nil {
|
|
handleError(errors.New("the field 'jobId' is required"), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
// log.Printf("loading db job for stopJobByRequest... : stopJobApiRequest=%v", req)
|
|
job, err = api.JobRepository.Find(req.JobId, req.Cluster, req.StartTime)
|
|
if err != nil {
|
|
handleError(fmt.Errorf("finding job failed: %w", err), http.StatusUnprocessableEntity, rw)
|
|
return
|
|
}
|
|
|
|
api.checkAndHandleStopJob(rw, job, req)
|
|
}
|
|
|
|
// deleteJobById godoc
|
|
// @summary Remove a job from the sql database
|
|
// @tags Job remove
|
|
// @description Job to remove is specified by database ID. This will not remove the job from the job archive.
|
|
// @produce json
|
|
// @param id path int true "Database ID of Job"
|
|
// @success 200 {object} api.DeleteJobApiResponse "Success message"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 404 {object} api.ErrorResponse "Resource not found"
|
|
// @failure 422 {object} api.ErrorResponse "Unprocessable Entity: finding job failed: sql: no rows in result set"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/delete_job/{id} [delete]
|
|
func (api *RestApi) deleteJobById(rw http.ResponseWriter, r *http.Request) {
|
|
// Fetch job (that will be stopped) from db
|
|
id, ok := mux.Vars(r)["id"]
|
|
var err error
|
|
if ok {
|
|
id, e := strconv.ParseInt(id, 10, 64)
|
|
if e != nil {
|
|
handleError(fmt.Errorf("integer expected in path for id: %w", e), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
err = api.JobRepository.DeleteJobById(id)
|
|
} else {
|
|
handleError(errors.New("the parameter 'id' is required"), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
if err != nil {
|
|
handleError(fmt.Errorf("deleting job failed: %w", err), http.StatusUnprocessableEntity, rw)
|
|
return
|
|
}
|
|
rw.Header().Add("Content-Type", "application/json")
|
|
rw.WriteHeader(http.StatusOK)
|
|
json.NewEncoder(rw).Encode(DeleteJobApiResponse{
|
|
Message: fmt.Sprintf("Successfully deleted job %s", id),
|
|
})
|
|
}
|
|
|
|
// deleteJobByRequest godoc
|
|
// @summary Remove a job from the sql database
|
|
// @tags Job remove
|
|
// @description Job to delete is specified by request body. All fields are required in this case.
|
|
// @accept json
|
|
// @produce json
|
|
// @param request body api.DeleteJobApiRequest true "All fields required"
|
|
// @success 200 {object} api.DeleteJobApiResponse "Success message"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 404 {object} api.ErrorResponse "Resource not found"
|
|
// @failure 422 {object} api.ErrorResponse "Unprocessable Entity: finding job failed: sql: no rows in result set"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/delete_job/ [delete]
|
|
func (api *RestApi) deleteJobByRequest(rw http.ResponseWriter, r *http.Request) {
|
|
// Parse request body
|
|
req := DeleteJobApiRequest{}
|
|
if err := decode(r.Body, &req); err != nil {
|
|
handleError(fmt.Errorf("parsing request body failed: %w", err), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
// Fetch job (that will be deleted) from db
|
|
var job *schema.Job
|
|
var err error
|
|
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)
|
|
if err != nil {
|
|
handleError(fmt.Errorf("finding job failed: %w", err), http.StatusUnprocessableEntity, rw)
|
|
return
|
|
}
|
|
|
|
err = api.JobRepository.DeleteJobById(job.ID)
|
|
if err != nil {
|
|
handleError(fmt.Errorf("deleting job failed: %w", err), http.StatusUnprocessableEntity, rw)
|
|
return
|
|
}
|
|
|
|
rw.Header().Add("Content-Type", "application/json")
|
|
rw.WriteHeader(http.StatusOK)
|
|
json.NewEncoder(rw).Encode(DeleteJobApiResponse{
|
|
Message: fmt.Sprintf("Successfully deleted job %d", job.ID),
|
|
})
|
|
}
|
|
|
|
// deleteJobBefore godoc
|
|
// @summary Remove a job from the sql database
|
|
// @tags Job remove
|
|
// @description Remove all jobs with start time before timestamp. The jobs will not be removed from the job archive.
|
|
// @produce json
|
|
// @param ts path int true "Unix epoch timestamp"
|
|
// @success 200 {object} api.DeleteJobApiResponse "Success message"
|
|
// @failure 400 {object} api.ErrorResponse "Bad Request"
|
|
// @failure 401 {object} api.ErrorResponse "Unauthorized"
|
|
// @failure 403 {object} api.ErrorResponse "Forbidden"
|
|
// @failure 404 {object} api.ErrorResponse "Resource not found"
|
|
// @failure 422 {object} api.ErrorResponse "Unprocessable Entity: finding job failed: sql: no rows in result set"
|
|
// @failure 500 {object} api.ErrorResponse "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /jobs/delete_job_before/{ts} [delete]
|
|
func (api *RestApi) deleteJobBefore(rw http.ResponseWriter, r *http.Request) {
|
|
var cnt int
|
|
// Fetch job (that will be stopped) from db
|
|
id, ok := mux.Vars(r)["ts"]
|
|
var err error
|
|
if ok {
|
|
ts, e := strconv.ParseInt(id, 10, 64)
|
|
if e != nil {
|
|
handleError(fmt.Errorf("integer expected in path for ts: %w", e), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
cnt, err = api.JobRepository.DeleteJobsBefore(ts)
|
|
} else {
|
|
handleError(errors.New("the parameter 'ts' is required"), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
if err != nil {
|
|
handleError(fmt.Errorf("deleting jobs failed: %w", err), http.StatusUnprocessableEntity, rw)
|
|
return
|
|
}
|
|
|
|
rw.Header().Add("Content-Type", "application/json")
|
|
rw.WriteHeader(http.StatusOK)
|
|
json.NewEncoder(rw).Encode(DeleteJobApiResponse{
|
|
Message: fmt.Sprintf("Successfully deleted %d jobs", cnt),
|
|
})
|
|
}
|
|
|
|
func (api *RestApi) checkAndHandleStopJob(rw http.ResponseWriter, job *schema.Job, req StopJobApiRequest) {
|
|
// Sanity checks
|
|
if job == nil || job.StartTime.Unix() >= req.StopTime || job.State != schema.JobStateRunning {
|
|
handleError(fmt.Errorf("jobId %d (id %d) on %s : stopTime %d must be larger than startTime %d and only running jobs can be stopped (state is: %s)", job.JobID, job.ID, job.Cluster, req.StopTime, job.StartTime.Unix(), job.State), http.StatusBadRequest, rw)
|
|
return
|
|
}
|
|
|
|
if req.State != "" && !req.State.Valid() {
|
|
handleError(fmt.Errorf("jobId %d (id %d) on %s : invalid requested job state: %#v", job.JobID, job.ID, job.Cluster, req.State), http.StatusBadRequest, rw)
|
|
return
|
|
} else if req.State == "" {
|
|
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("jobId %d (id %d) on %s : marking job as '%s' (duration: %d) in DB failed: %w", job.JobID, job.ID, job.Cluster, job.State, job.Duration, err), http.StatusInternalServerError, rw)
|
|
return
|
|
}
|
|
|
|
log.Printf("archiving job... (dbid: %d): cluster=%s, jobId=%d, user=%s, startTime=%s, duration=%d, state=%s", job.ID, job.Cluster, job.JobID, job.User, job.StartTime, job.Duration, job.State)
|
|
|
|
// 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
|
|
}
|
|
|
|
// Trigger async archiving
|
|
archiver.TriggerArchiving(job)
|
|
}
|
|
|
|
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"`
|
|
}
|
|
|
|
resolver := graph.GetResolverInstance()
|
|
data, err := resolver.Query().JobMetrics(r.Context(), id, metrics, scopes, nil)
|
|
if err != nil {
|
|
json.NewEncoder(rw).Encode(Respone{
|
|
Error: &struct {
|
|
Message string "json:\"message\""
|
|
}{Message: err.Error()},
|
|
})
|
|
return
|
|
}
|
|
|
|
json.NewEncoder(rw).Encode(Respone{
|
|
Data: &struct {
|
|
JobMetrics []*model.JobMetricWithName "json:\"jobMetrics\""
|
|
}{JobMetrics: data},
|
|
})
|
|
}
|
|
|
|
// createUser godoc
|
|
// @summary Adds a new user
|
|
// @tags User
|
|
// @description User specified in form data will be saved to database.
|
|
// @description Only accessible from IPs registered with apiAllowedIPs configuration option.
|
|
// @accept mpfd
|
|
// @produce plain
|
|
// @param username formData string true "Unique user ID"
|
|
// @param password formData string true "User password"
|
|
// @param role formData string true "User role" Enums(admin, support, manager, user, api)
|
|
// @param project formData string false "Managed project, required for new manager role user"
|
|
// @param name formData string false "Users name"
|
|
// @param email formData string false "Users email"
|
|
// @success 200 {string} string "Success Response"
|
|
// @failure 400 {string} string "Bad Request"
|
|
// @failure 401 {string} string "Unauthorized"
|
|
// @failure 403 {string} string "Forbidden"
|
|
// @failure 422 {string} string "Unprocessable Entity: creating user failed"
|
|
// @failure 500 {string} string "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /users/ [post]
|
|
func (api *RestApi) createUser(rw http.ResponseWriter, r *http.Request) {
|
|
err := securedCheck(r)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
rw.Header().Set("Content-Type", "text/plain")
|
|
me := repository.GetUserFromContext(r.Context())
|
|
if !me.HasRole(schema.RoleAdmin) {
|
|
http.Error(rw, "Only admins are allowed to create new users", http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
username, password, role, name, email, project := r.FormValue("username"),
|
|
r.FormValue("password"), r.FormValue("role"), r.FormValue("name"),
|
|
r.FormValue("email"), r.FormValue("project")
|
|
|
|
if len(password) == 0 && role != schema.GetRoleString(schema.RoleApi) {
|
|
http.Error(rw, "Only API users are allowed to have a blank password (login will be impossible)", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if len(project) != 0 && role != schema.GetRoleString(schema.RoleManager) {
|
|
http.Error(rw, "only managers require a project (can be changed later)",
|
|
http.StatusBadRequest)
|
|
return
|
|
} else if len(project) == 0 && role == schema.GetRoleString(schema.RoleManager) {
|
|
http.Error(rw, "managers require a project to manage (can be changed later)",
|
|
http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
if err := repository.GetUserRepository().AddUser(&schema.User{
|
|
Username: username,
|
|
Name: name,
|
|
Password: password,
|
|
Email: email,
|
|
Projects: []string{project},
|
|
Roles: []string{role},
|
|
}); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
|
|
fmt.Fprintf(rw, "User %v successfully created!\n", username)
|
|
}
|
|
|
|
// deleteUser godoc
|
|
// @summary Deletes a user
|
|
// @tags User
|
|
// @description User defined by username in form data will be deleted from database.
|
|
// @description Only accessible from IPs registered with apiAllowedIPs configuration option.
|
|
// @accept mpfd
|
|
// @produce plain
|
|
// @param username formData string true "User ID to delete"
|
|
// @success 200 "User deleted successfully"
|
|
// @failure 400 {string} string "Bad Request"
|
|
// @failure 401 {string} string "Unauthorized"
|
|
// @failure 403 {string} string "Forbidden"
|
|
// @failure 422 {string} string "Unprocessable Entity: deleting user failed"
|
|
// @failure 500 {string} string "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /users/ [delete]
|
|
func (api *RestApi) deleteUser(rw http.ResponseWriter, r *http.Request) {
|
|
err := securedCheck(r)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
if user := repository.GetUserFromContext(r.Context()); !user.HasRole(schema.RoleAdmin) {
|
|
http.Error(rw, "Only admins are allowed to delete a user", http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
username := r.FormValue("username")
|
|
if err := repository.GetUserRepository().DelUser(username); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
|
|
rw.WriteHeader(http.StatusOK)
|
|
}
|
|
|
|
// getUsers godoc
|
|
// @summary Returns a list of users
|
|
// @tags User
|
|
// @description Returns a JSON-encoded list of users.
|
|
// @description Required query-parameter defines if all users or only users with additional special roles are returned.
|
|
// @description Only accessible from IPs registered with apiAllowedIPs configuration option.
|
|
// @produce json
|
|
// @param not-just-user query bool true "If returned list should contain all users or only users with additional special roles"
|
|
// @success 200 {array} api.ApiReturnedUser "List of users returned successfully"
|
|
// @failure 400 {string} string "Bad Request"
|
|
// @failure 401 {string} string "Unauthorized"
|
|
// @failure 403 {string} string "Forbidden"
|
|
// @failure 500 {string} string "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /users/ [get]
|
|
func (api *RestApi) getUsers(rw http.ResponseWriter, r *http.Request) {
|
|
err := securedCheck(r)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
if user := repository.GetUserFromContext(r.Context()); !user.HasRole(schema.RoleAdmin) {
|
|
http.Error(rw, "Only admins are allowed to fetch a list of users", http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
users, err := repository.GetUserRepository().ListUsers(r.URL.Query().Get("not-just-user") == "true")
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
json.NewEncoder(rw).Encode(users)
|
|
}
|
|
|
|
// updateUser godoc
|
|
// @summary Updates an existing user
|
|
// @tags User
|
|
// @description Modifies user defined by username (id) in one of four possible ways.
|
|
// @description If more than one formValue is set then only the highest priority field is used.
|
|
// @description Only accessible from IPs registered with apiAllowedIPs configuration option.
|
|
// @accept mpfd
|
|
// @produce plain
|
|
// @param id path string true "Database ID of User"
|
|
// @param add-role formData string false "Priority 1: Role to add" Enums(admin, support, manager, user, api)
|
|
// @param remove-role formData string false "Priority 2: Role to remove" Enums(admin, support, manager, user, api)
|
|
// @param add-project formData string false "Priority 3: Project to add"
|
|
// @param remove-project formData string false "Priority 4: Project to remove"
|
|
// @success 200 {string} string "Success Response Message"
|
|
// @failure 400 {string} string "Bad Request"
|
|
// @failure 401 {string} string "Unauthorized"
|
|
// @failure 403 {string} string "Forbidden"
|
|
// @failure 422 {string} string "Unprocessable Entity: The user could not be updated"
|
|
// @failure 500 {string} string "Internal Server Error"
|
|
// @security ApiKeyAuth
|
|
// @router /user/{id} [post]
|
|
func (api *RestApi) updateUser(rw http.ResponseWriter, r *http.Request) {
|
|
err := securedCheck(r)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
if user := repository.GetUserFromContext(r.Context()); !user.HasRole(schema.RoleAdmin) {
|
|
http.Error(rw, "Only admins are allowed to update a user", http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
// Get Values
|
|
newrole := r.FormValue("add-role")
|
|
delrole := r.FormValue("remove-role")
|
|
newproj := r.FormValue("add-project")
|
|
delproj := r.FormValue("remove-project")
|
|
|
|
// TODO: Handle anything but roles...
|
|
if newrole != "" {
|
|
if err := repository.GetUserRepository().AddRole(r.Context(), mux.Vars(r)["id"], newrole); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
rw.Write([]byte("Add Role Success"))
|
|
} else if delrole != "" {
|
|
if err := repository.GetUserRepository().RemoveRole(r.Context(), mux.Vars(r)["id"], delrole); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
rw.Write([]byte("Remove Role Success"))
|
|
} else if newproj != "" {
|
|
if err := repository.GetUserRepository().AddProject(r.Context(), mux.Vars(r)["id"], newproj); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
rw.Write([]byte("Add Project Success"))
|
|
} else if delproj != "" {
|
|
if err := repository.GetUserRepository().RemoveProject(r.Context(), mux.Vars(r)["id"], delproj); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
rw.Write([]byte("Remove Project Success"))
|
|
} else {
|
|
http.Error(rw, "Not Add or Del [role|project]?", http.StatusInternalServerError)
|
|
}
|
|
}
|
|
|
|
func (api *RestApi) getJWT(rw http.ResponseWriter, r *http.Request) {
|
|
err := securedCheck(r)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
rw.Header().Set("Content-Type", "text/plain")
|
|
username := r.FormValue("username")
|
|
me := repository.GetUserFromContext(r.Context())
|
|
if !me.HasRole(schema.RoleAdmin) {
|
|
if username != me.Username {
|
|
http.Error(rw, "Only admins are allowed to sign JWTs not for themselves",
|
|
http.StatusForbidden)
|
|
return
|
|
}
|
|
}
|
|
|
|
user, err := repository.GetUserRepository().GetUser(username)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
|
|
jwt, err := api.Authentication.JwtAuth.ProvideJWT(user)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
|
|
rw.WriteHeader(http.StatusOK)
|
|
rw.Write([]byte(jwt))
|
|
}
|
|
|
|
func (api *RestApi) getRoles(rw http.ResponseWriter, r *http.Request) {
|
|
err := securedCheck(r)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
user := repository.GetUserFromContext(r.Context())
|
|
if !user.HasRole(schema.RoleAdmin) {
|
|
http.Error(rw, "only admins are allowed to fetch a list of roles", http.StatusForbidden)
|
|
return
|
|
}
|
|
|
|
roles, err := schema.GetValidRoles(user)
|
|
if err != nil {
|
|
http.Error(rw, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
json.NewEncoder(rw).Encode(roles)
|
|
}
|
|
|
|
func (api *RestApi) updateConfiguration(rw http.ResponseWriter, r *http.Request) {
|
|
rw.Header().Set("Content-Type", "text/plain")
|
|
key, value := r.FormValue("key"), r.FormValue("value")
|
|
|
|
fmt.Printf("REST > KEY: %#v\nVALUE: %#v\n", key, value)
|
|
|
|
if err := repository.GetUserCfgRepo().UpdateConfig(key, value, repository.GetUserFromContext(r.Context())); err != nil {
|
|
http.Error(rw, err.Error(), http.StatusUnprocessableEntity)
|
|
return
|
|
}
|
|
|
|
rw.Write([]byte("success"))
|
|
}
|
|
|
|
func (api *RestApi) putMachineState(rw http.ResponseWriter, r *http.Request) {
|
|
if api.MachineStateDir == "" {
|
|
http.Error(rw, "REST > machine state 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, "REST > machine state 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)
|
|
}
|