Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add job api #751

Merged
merged 14 commits into from
Oct 26, 2021
Prev Previous commit
Next Next commit
feat: job
Signed-off-by: Gaius <gaius.qi@gmail.com>
gaius-qi committed Oct 22, 2021
commit ecf337217fd308cced0320f6c6fa98f7f49a1d6e
185 changes: 169 additions & 16 deletions manager/handlers/job.go
Original file line number Diff line number Diff line change
@@ -1,17 +1,170 @@
/*
* Copyright 2020 The Dragonfly Authors
*
* 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 handlers

import (
"net/http"

"d7y.io/dragonfly/v2/internal/job"
"d7y.io/dragonfly/v2/manager/types"
"github.com/gin-gonic/gin"
)

// @Summary Create Job
// @Description create by json config
// @Tags Job
// @Accept json
// @Produce json
// @Param Job body types.CreateJobRequest true "Job"
// @Success 200 {object} model.Job
// @Failure 400
// @Failure 404
// @Failure 500
// @Router /jobs [post]
func (h *Handlers) CreateJob(ctx *gin.Context) {
var json types.CreateJobRequest
if err := ctx.ShouldBindJSON(&json); err != nil {
ctx.JSON(http.StatusUnprocessableEntity, gin.H{"errors": err.Error()})
return
}

switch json.Type {
case job.PreheatJob:
var json types.CreatePreheatRequest
if err := ctx.ShouldBindJSON(&json); err != nil {
ctx.JSON(http.StatusUnprocessableEntity, gin.H{"errors": err.Error()})
return
}

job, err := h.service.CreatePreheatJob(ctx.Request.Context(), json)
if err != nil {
ctx.Error(err)
return
}

ctx.JSON(http.StatusOK, job)
default:
ctx.JSON(http.StatusUnprocessableEntity, gin.H{"errors": "Unknow type"})
}
}

// @Summary Destroy Job
// @Description Destroy by id
// @Tags Job
// @Accept json
// @Produce json
// @Param id path string true "id"
// @Success 200
// @Failure 400
// @Failure 404
// @Failure 500
// @Router /jobs/{id} [delete]
func (h *Handlers) DestroyJob(ctx *gin.Context) {
var params types.JobParams
if err := ctx.ShouldBindUri(&params); err != nil {
ctx.JSON(http.StatusUnprocessableEntity, gin.H{"errors": err.Error()})
return
}

if err := h.service.DestroyJob(ctx.Request.Context(), params.ID); err != nil {
ctx.Error(err)
return
}

ctx.Status(http.StatusOK)
}

// @Summary Update Job
// @Description Update by json config
// @Tags Job
// @Accept json
// @Produce json
// @Param id path string true "id"
// @Param Job body types.UpdateJobRequest true "Job"
// @Success 200 {object} model.Job
// @Failure 400
// @Failure 404
// @Failure 500
// @Router /jobs/{id} [patch]
func (h *Handlers) UpdateJob(ctx *gin.Context) {
var params types.JobParams
if err := ctx.ShouldBindUri(&params); err != nil {
ctx.Error(err)
return
}

var json types.UpdateJobRequest
if err := ctx.ShouldBindJSON(&json); err != nil {
ctx.Error(err)
return
}

job, err := h.service.UpdateJob(ctx.Request.Context(), params.ID, json)
if err != nil {
ctx.Error(err)
return
}

ctx.JSON(http.StatusOK, job)
}

// @Summary Get Job
// @Description Get Job by id
// @Tags Job
// @Accept json
// @Produce json
// @Param id path string true "id"
// @Success 200 {object} model.Job
// @Failure 400
// @Failure 404
// @Failure 500
// @Router /jobs/{id} [get]
func (h *Handlers) GetJob(ctx *gin.Context) {
var params types.JobParams
if err := ctx.ShouldBindUri(&params); err != nil {
ctx.JSON(http.StatusUnprocessableEntity, gin.H{"errors": err.Error()})
return
}

job, err := h.service.GetJob(ctx.Request.Context(), params.ID)
if err != nil {
ctx.Error(err)
return
}

ctx.JSON(http.StatusOK, job)
}

// @Summary Get Jobs
// @Description Get Jobs
// @Tags Job
// @Accept json
// @Produce json
// @Param page query int true "current page" default(0)
// @Param per_page query int true "return max item count, default 10, max 50" default(10) minimum(2) maximum(50)
// @Success 200 {object} []model.Job
// @Failure 400
// @Failure 404
// @Failure 500
// @Router /jobs [get]
func (h *Handlers) GetJobs(ctx *gin.Context) {
var query types.GetJobsQuery
if err := ctx.ShouldBindQuery(&query); err != nil {
ctx.JSON(http.StatusUnprocessableEntity, gin.H{"errors": err.Error()})
return
}

h.setPaginationDefault(&query.Page, &query.PerPage)
jobs, err := h.service.GetJobs(ctx.Request.Context(), query)
if err != nil {
ctx.Error(err)
return
}

totalCount, err := h.service.JobTotalCount(ctx.Request.Context(), query)
if err != nil {
ctx.Error(err)
return
}

h.setPaginationLinkHeader(ctx, query.Page, query.PerPage, int(totalCount))
ctx.JSON(http.StatusOK, jobs)
}
22 changes: 11 additions & 11 deletions manager/job/preheat.go
Original file line number Diff line number Diff line change
@@ -52,8 +52,8 @@ const (
var accessURLPattern, _ = regexp.Compile("^(.*)://(.*)/v2/(.*)/manifests/(.*)")

type Preheat interface {
CreatePreheat(context.Context, []model.Scheduler, types.CreatePreheatRequest) (*types.Preheat, error)
GetPreheat(context.Context, string) (*types.Preheat, error)
CreatePreheat(context.Context, []model.Scheduler, types.PreheatArgs) (*internaljob.GroupJobState, error)
GetPreheat(context.Context, string) (*internaljob.GroupJobState, error)
}

type preheat struct {
@@ -75,20 +75,20 @@ func newPreheat(job *internaljob.Job, bizTag string) (Preheat, error) {
}, nil
}

func (p *preheat) GetPreheat(ctx context.Context, id string) (*types.Preheat, error) {
func (p *preheat) GetPreheat(ctx context.Context, id string) (*internaljob.GroupJobState, error) {
groupJobState, err := p.job.GetGroupJobState(id)
if err != nil {
return nil, err
}

return &types.Preheat{
ID: groupJobState.GroupUUID,
Status: groupJobState.State,
return &internaljob.GroupJobState{
GroupUUID: groupJobState.GroupUUID,
State: groupJobState.State,
CreatedAt: groupJobState.CreatedAt,
}, nil
}

func (p *preheat) CreatePreheat(ctx context.Context, schedulers []model.Scheduler, json types.CreatePreheatRequest) (*types.Preheat, error) {
func (p *preheat) CreatePreheat(ctx context.Context, schedulers []model.Scheduler, json types.PreheatArgs) (*internaljob.GroupJobState, error) {
var span trace.Span
ctx, span = tracer.Start(ctx, config.SpanPreheat, trace.WithSpanKind(trace.SpanKindProducer))
span.SetAttributes(config.AttributePreheatType.String(json.Type))
@@ -136,7 +136,7 @@ func (p *preheat) CreatePreheat(ctx context.Context, schedulers []model.Schedule
return p.createGroupJob(ctx, files, queues)
}

func (p *preheat) createGroupJob(ctx context.Context, files []*internaljob.PreheatRequest, queues []internaljob.Queue) (*types.Preheat, error) {
func (p *preheat) createGroupJob(ctx context.Context, files []*internaljob.PreheatRequest, queues []internaljob.Queue) (*internaljob.GroupJobState, error) {
signatures := []*machineryv1tasks.Signature{}
var urls []string
for i := range files {
@@ -169,9 +169,9 @@ func (p *preheat) createGroupJob(ctx context.Context, files []*internaljob.Prehe
}

logger.Infof("create preheat group job successed, group uuid: %s, urls:%s", group.GroupUUID, urls)
return &types.Preheat{
ID: group.GroupUUID,
Status: machineryv1tasks.StatePending,
return &internaljob.GroupJobState{
GroupUUID: group.GroupUUID,
State: machineryv1tasks.StatePending,
CreatedAt: time.Now(),
}, nil
}
3 changes: 2 additions & 1 deletion manager/model/job.go
Original file line number Diff line number Diff line change
@@ -22,7 +22,8 @@ type Job struct {
BIO string `gorm:"column:bio;type:varchar(1024);comment:biography" json:"bio"`
Type string `gorm:"column:type;type:varchar(256);comment:type" json:"type"`
Status string `gorm:"column:status;type:varchar(256);default:'PENDING';comment:service status" json:"status"`
Body JSONMap `gorm:"column:body;not null;comment:task request body" json:"body"`
Args JSONMap `gorm:"column:args;not null;comment:task request args" json:"args"`
Result JSONMap `gorm:"column:result;not null;comment:task result" json:"result"`
UserID uint `gorm:"comment:user id" json:"user_id"`
User User `json:"-"`
CDNClusters []CDNCluster `gorm:"many2many:job_cdn_cluster;" json:"cdn_clusters"`
8 changes: 8 additions & 0 deletions manager/router/router.go
Original file line number Diff line number Diff line change
@@ -182,6 +182,14 @@ func Init(cfg *config.Config, service service.REST, enforcer *casbin.Enforcer) (
config.GET(":id", h.GetConfig)
config.GET("", h.GetConfigs)

// Job
job := apiv1.Group("/jobs")
job.POST("", h.CreateJob)
job.DELETE(":id", h.DestroyJob)
job.PATCH(":id", h.UpdateJob)
job.GET(":id", h.GetJob)
job.GET("", h.GetJobs)

// Preheat
ph := apiv1.Group("/preheats")
ph.POST("", h.CreatePreheat)
Loading