Upstream
/
harbor
mirror of https://github.com/goharbor/harbor.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

add extra attributes in the schedule table 

Signed-off-by: Wang Yan <wangyan@vmware.com>
This commit is contained in:
Wang Yan 2020-12-11 17:45:41 +08:00
parent 1bb79d402d
commit 39e1a4f2b4
17 changed files with 105 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
make/migrations/postgresql/0050_2.2.0_schema.up.sql
View File

@ -268,6 +268,7 @@ BEGIN
UPDATE scanner_registration SET is_default = TRUE WHERE name = 'Trivy' AND immutable = TRUE;
END IF;
END $$;
ALTER TABLE execution ALTER COLUMN vendor_type TYPE varchar(64);
ALTER TABLE schedule ALTER COLUMN vendor_type TYPE varchar(64);
ALTER TABLE task ALTER COLUMN vendor_type TYPE varchar(64);
ALTER TABLE execution ALTER COLUMN vendor_type type varchar(64);
ALTER TABLE schedule ALTER COLUMN vendor_type type varchar(64);
ALTER TABLE schedule ADD COLUMN IF NOT EXISTS extra_attrs JSON;
ALTER TABLE task ALTER COLUMN vendor_type type varchar(64);

27
src/controller/gc/callback.go Normal file
View File

@ -0,0 +1,27 @@
package gc
import (
"context"
"encoding/json"
"fmt"
"github.com/goharbor/harbor/src/lib/log"
"github.com/goharbor/harbor/src/lib/orm"
"github.com/goharbor/harbor/src/pkg/scheduler"
"github.com/goharbor/harbor/src/pkg/task"
)
func init() {
err := scheduler.RegisterCallbackFunc(SchedulerCallback, gcCallback)
if err != nil {
log.Fatalf("failed to registry GC call back, %v", err)
}
}
func gcCallback(ctx context.Context, p string) error {
param := &Policy{}
if err := json.Unmarshal([]byte(p), param); err != nil {
return fmt.Errorf("failed to unmarshal the param: %v", err)
}
_, err := Ctl.Start(orm.Context(), *param, task.ExecutionTriggerSchedule)
return err
}

4
src/controller/gc/controller.go
View File

@ -177,7 +177,9 @@ func (c *controller) GetSchedule(ctx context.Context) (*scheduler.Schedule, erro
// CreateSchedule ...
func (c *controller) CreateSchedule(ctx context.Context, cronType, cron string, policy Policy) (int64, error) {
return c.schedulerMgr.Schedule(ctx, GCVendorType, -1, cronType, cron, SchedulerCallback, policy)
extras := make(map[string]interface{})
extras["delete_untagged"] = policy.DeleteUntagged
return c.schedulerMgr.Schedule(ctx, GCVendorType, -1, cronType, cron, SchedulerCallback, policy, extras)
}
// DeleteSchedule ...

2
src/controller/gc/controller_test.go
View File

@ -126,7 +126,7 @@ func (g *gcCtrTestSuite) TestGetSchedule() {
func (g *gcCtrTestSuite) TestCreateSchedule() {
g.scheduler.On("Schedule", mock.Anything, mock.Anything, mock.Anything, mock.Anything,
mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
dataMap := make(map[string]interface{})
p := Policy{

6
src/controller/p2p/preheat/controller.go
View File

@ -292,8 +292,9 @@ func (c *controller) CreatePolicy(ctx context.Context, schema *policyModels.Sche
schema.Trigger.Type == policyModels.TriggerTypeScheduled &&
len(schema.Trigger.Settings.Cron) > 0 {
// schedule and update policy
extras := make(map[string]interface{})
if _, err = c.scheduler.Schedule(ctx, job.P2PPreheat, id, "", schema.Trigger.Settings.Cron,
SchedulerCallback, TriggerParam{PolicyID: id}); err != nil {
SchedulerCallback, TriggerParam{PolicyID: id}, extras); err != nil {
return 0, err
}
@ -384,8 +385,9 @@ func (c *controller) UpdatePolicy(ctx context.Context, schema *policyModels.Sche
// schedule new
if needSch {
extras := make(map[string]interface{})
if _, err := c.scheduler.Schedule(ctx, job.P2PPreheat, schema.ID, "", cron, SchedulerCallback,
TriggerParam{PolicyID: schema.ID}); err != nil {
TriggerParam{PolicyID: schema.ID}, extras); err != nil {
return err
}
}

2
src/controller/p2p/preheat/controllor_test.go
View File

@ -241,7 +241,7 @@ func (s *preheatSuite) TestCreatePolicy() {
FiltersStr: `[{"type":"repository","value":"harbor*"},{"type":"tag","value":"2*"}]`,
TriggerStr: fmt.Sprintf(`{"type":"%s", "trigger_setting":{"cron":"* * * * */1"}}`, policy.TriggerTypeScheduled),
}
s.fakeScheduler.On("Schedule", s.ctx, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
s.fakeScheduler.On("Schedule", s.ctx, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
s.fakePolicyMgr.On("Create", s.ctx, policy).Return(int64(1), nil)
s.fakePolicyMgr.On("Update", s.ctx, mock.Anything, mock.Anything).Return(nil)
s.fakeScheduler.On("UnScheduleByVendor", s.ctx, mock.Anything, mock.Anything).Return(nil)

16
src/core/api/base.go
View File

@ -18,8 +18,6 @@ import (
"context"
"encoding/json"
"fmt"
"github.com/goharbor/harbor/src/lib/orm"
"github.com/goharbor/harbor/src/pkg/task"
"net/http"
"github.com/ghodss/yaml"
@ -28,7 +26,6 @@ import (
"github.com/goharbor/harbor/src/common/rbac"
"github.com/goharbor/harbor/src/common/security"
"github.com/goharbor/harbor/src/common/utils"
"github.com/goharbor/harbor/src/controller/gc"
"github.com/goharbor/harbor/src/controller/p2p/preheat"
projectcontroller "github.com/goharbor/harbor/src/controller/project"
"github.com/goharbor/harbor/src/core/config"
@ -52,7 +49,6 @@ var (
retentionMgr retention.Manager
retentionLauncher retention.Launcher
retentionController retention.APIController
gcController gc.Controller
)
// GetRetentionController returns the retention API controller
@ -198,8 +194,6 @@ func Init() error {
retentionController = retention.NewAPIController(retentionMgr, projectMgr, repository.Mgr, scheduler.Sched, retentionLauncher)
gcController = gc.NewController()
retentionCallbackFun := func(ctx context.Context, p string) error {
param := &retention.TriggerParam{}
if err := json.Unmarshal([]byte(p), param); err != nil {
@ -223,16 +217,6 @@ func Init() error {
}
err = scheduler.RegisterCallbackFunc(preheat.SchedulerCallback, p2pPreheatCallbackFun)
gcCallbackFun := func(ctx context.Context, p string) error {
param := &gc.Policy{}
if err := json.Unmarshal([]byte(p), param); err != nil {
return fmt.Errorf("failed to unmarshal the param: %v", err)
}
_, err := gcController.Start(orm.Context(), *param, task.ExecutionTriggerSchedule)
return err
}
err = scheduler.RegisterCallbackFunc(gc.SchedulerCallback, gcCallbackFun)
return err
}

6
src/pkg/retention/controller.go
View File

@ -93,10 +93,11 @@ func (r *DefaultAPIController) CreateRetention(p *policy.Metadata) (int64, error
if p.Trigger.Kind == policy.TriggerKindSchedule {
cron, ok := p.Trigger.Settings[policy.TriggerSettingsCron]
if ok && len(cron.(string)) > 0 {
extras := make(map[string]interface{})
if _, err = r.scheduler.Schedule(orm.Context(), schedulerVendorType, id, "", cron.(string), SchedulerCallback, TriggerParam{
PolicyID: id,
Trigger: ExecutionTriggerSchedule,
}); err != nil {
}, extras); err != nil {
return 0, err
}
}
@ -152,10 +153,11 @@ func (r *DefaultAPIController) UpdateRetention(p *policy.Metadata) error {
}
}
if needSch {
extras := make(map[string]interface{})
_, err := r.scheduler.Schedule(orm.Context(), schedulerVendorType, p.ID, "", p.Trigger.Settings[policy.TriggerSettingsCron].(string), SchedulerCallback, TriggerParam{
PolicyID: p.ID,
Trigger: ExecutionTriggerSchedule,
})
}, extras)
if err != nil {
return err
}

2
src/pkg/retention/controller_test.go
View File

@ -220,7 +220,7 @@ func (s *ControllerTestSuite) TestExecution() {
type fakeRetentionScheduler struct {
}
func (f *fakeRetentionScheduler) Schedule(ctx context.Context, vendorType string, vendorID int64, cronType string, cron string, callbackFuncName string, params interface{}) (int64, error) {
func (f *fakeRetentionScheduler) Schedule(ctx context.Context, vendorType string, vendorID int64, cronType string, cron string, callbackFuncName string, params interface{}, extras map[string]interface{}) (int64, error) {
return 111, nil
}

1
src/pkg/scheduler/dao.go
View File

@ -34,6 +34,7 @@ type schedule struct {
VendorID int64 `orm:"column(vendor_id)"`
CRONType string `orm:"column(cron_type)"`
CRON string `orm:"column(cron)"`
ExtraAttrs string `orm:"column(extra_attrs)"`
CallbackFuncName string `orm:"column(callback_func_name)"`
CallbackFuncParam string `orm:"column(callback_func_param)"`
CreationTime time.Time `orm:"column(creation_time)"`

4
src/pkg/scheduler/dao_test.go
View File

@ -45,6 +45,7 @@ func (d *daoTestSuite) SetupTest() {
CRON: "0 * * * * *",
CallbackFuncName: "callback_func_01",
CallbackFuncParam: "callback_func_params",
ExtraAttrs: `{"key":"value"}`,
}
id, err := d.dao.Create(d.ctx, schedule)
d.Require().Nil(err)
@ -79,6 +80,7 @@ func (d *daoTestSuite) TestGet() {
schedule, err = d.dao.Get(d.ctx, d.id)
d.Require().Nil(err)
d.Equal(d.id, schedule.ID)
d.Equal("{\"key\":\"value\"}", schedule.ExtraAttrs)
}
func (d *daoTestSuite) TestDelete() {
@ -93,7 +95,7 @@ func (d *daoTestSuite) TestUpdate() {
// not found
err := d.dao.Update(d.ctx, &schedule{
ID: 10000,
})
}, "CRON")
d.True(errors.IsNotFoundErr(err))
// pass

43
src/pkg/scheduler/scheduler.go
View File

@ -37,15 +37,15 @@ var (
// Schedule describes the detail information about the created schedule
type Schedule struct {
ID int64 `json:"id"`
VendorType string `json:"vendor_type"`
VendorID int64 `json:"vendor_id"`
CRONType string `json:"cron_type"`
CRON string `json:"cron"`
Param string `json:"param"`
Status string `json:"status"` // status of the underlying task(jobservice job)
CreationTime time.Time `json:"creation_time"`
UpdateTime time.Time `json:"update_time"`
ID int64 `json:"id"`
VendorType string `json:"vendor_type"`
VendorID int64 `json:"vendor_id"`
CRONType string `json:"cron_type"`
CRON string `json:"cron"`
ExtraAttrs map[string]interface{} `json:"extra_attrs"`
Status string `json:"status"` // status of the underlying task(jobservice job)
CreationTime time.Time `json:"creation_time"`
UpdateTime time.Time `json:"update_time"`
// we can extend this model to include more information(e.g. how many times the schedule already
// runs; when will the schedule runs next time)
}
@ -60,7 +60,7 @@ type Scheduler interface {
// The "params" is passed to the callback function as encoded json string, so the callback
// function must decode it before using
Schedule(ctx context.Context, vendorType string, vendorID int64, cronType string,
cron string, callbackFuncName string, params interface{}) (int64, error)
cron string, callbackFuncName string, params interface{}, extras map[string]interface{}) (int64, error)
// UnScheduleByID the schedule specified by ID
UnScheduleByID(ctx context.Context, id int64) error
// UnScheduleByVendor the schedule specified by vendor
@ -95,10 +95,10 @@ type scheduler struct {
// to out of control from the global transaction, and uses a new transaction that only
// covers the logic inside the function
func (s *scheduler) Schedule(ctx context.Context, vendorType string, vendorID int64, cronType string,
cron string, callbackFuncName string, params interface{}) (int64, error) {
cron string, callbackFuncName string, params interface{}, extras map[string]interface{}) (int64, error) {
var scheduleID int64
f := func(ctx context.Context) error {
id, err := s.schedule(ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params)
id, err := s.schedule(ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params, extras)
if err != nil {
return err
}
@ -114,7 +114,7 @@ func (s *scheduler) Schedule(ctx context.Context, vendorType string, vendorID in
}
func (s *scheduler) schedule(ctx context.Context, vendorType string, vendorID int64, cronType string,
cron string, callbackFuncName string, params interface{}) (int64, error) {
cron string, callbackFuncName string, params interface{}, extras map[string]interface{}) (int64, error) {
if len(vendorType) == 0 {
return 0, fmt.Errorf("empty vendor type")
}
@ -143,6 +143,13 @@ func (s *scheduler) schedule(ctx context.Context, vendorType string, vendorID in
}
sched.CallbackFuncParam = string(paramsData)
}
if extras != nil {
extrasData, err := json.Marshal(extras)
if err != nil {
return 0, err
}
sched.ExtraAttrs = string(extrasData)
}
// create schedule record
// when checkin hook comes, the database record must exist,
// so the database record must be created first before submitting job
@ -270,10 +277,18 @@ func (s *scheduler) convertSchedule(ctx context.Context, schedule *schedule) (*S
VendorID: schedule.VendorID,
CRONType: schedule.CRONType,
CRON: schedule.CRON,
Param: schedule.CallbackFuncParam,
CreationTime: schedule.CreationTime,
UpdateTime: schedule.UpdateTime,
}
if len(schedule.ExtraAttrs) > 0 {
extras := map[string]interface{}{}
if err := json.Unmarshal([]byte(schedule.ExtraAttrs), &extras); err != nil {
log.Errorf("failed to unmarshal the extra attributes of schedule %d: %v", schedule.ID, err)
return nil, err
}
schd.ExtraAttrs = extras
}
executions, err := s.execMgr.List(ctx, &q.Query{
Keywords: map[string]interface{}{
"VendorType": JobNameScheduler,

11
src/pkg/scheduler/scheduler_test.go
View File

@ -52,15 +52,16 @@ func (s *schedulerTestSuite) SetupTest() {
func (s *schedulerTestSuite) TestSchedule() {
// empty vendor type
id, err := s.scheduler.Schedule(nil, "", 0, "", "0 * * * * *", "callback", nil)
extras := make(map[string]interface{})
id, err := s.scheduler.Schedule(nil, "", 0, "", "0 * * * * *", "callback", nil, extras)
s.NotNil(err)
// invalid cron
id, err = s.scheduler.Schedule(nil, "vendor", 1, "", "", "callback", nil)
id, err = s.scheduler.Schedule(nil, "vendor", 1, "", "", "callback", nil, extras)
s.NotNil(err)
// callback function not exist
id, err = s.scheduler.Schedule(nil, "vendor", 1, "", "0 * * * * *", "not-exist", nil)
id, err = s.scheduler.Schedule(nil, "vendor", 1, "", "0 * * * * *", "not-exist", nil, extras)
s.NotNil(err)
// failed to submit to jobservice
@ -73,7 +74,7 @@ func (s *schedulerTestSuite) TestSchedule() {
Status: job.ErrorStatus.String(),
}, nil)
s.taskMgr.On("Stop", mock.Anything, mock.Anything).Return(nil)
_, err = s.scheduler.Schedule(nil, "vendor", 1, "", "0 * * * * *", "callback", "param")
_, err = s.scheduler.Schedule(nil, "vendor", 1, "", "0 * * * * *", "callback", "param", extras)
s.Require().NotNil(err)
s.dao.AssertExpectations(s.T())
s.execMgr.AssertExpectations(s.T())
@ -91,7 +92,7 @@ func (s *schedulerTestSuite) TestSchedule() {
ExecutionID: 1,
Status: job.SuccessStatus.String(),
}, nil)
id, err = s.scheduler.Schedule(nil, "vendor", 1, "", "0 * * * * *", "callback", "param")
id, err = s.scheduler.Schedule(nil, "vendor", 1, "", "0 * * * * *", "callback", "param", extras)
s.Require().Nil(err)
s.Equal(int64(1), id)
s.dao.AssertExpectations(s.T())

6
src/replication/policy/controller/controller.go
View File

@ -52,7 +52,8 @@ func (c *controller) Create(policy *model.Policy) (int64, error) {
return 0, err
}
if isScheduledTrigger(policy) {
if _, err = c.scheduler.Schedule(orm.Context(), job.Replication, id, "", policy.Trigger.Settings.Cron, CallbackFuncName, id); err != nil {
extras := make(map[string]interface{})
if _, err = c.scheduler.Schedule(orm.Context(), job.Replication, id, "", policy.Trigger.Settings.Cron, CallbackFuncName, id, extras); err != nil {
log.Errorf("failed to schedule the policy %d: %v", id, err)
}
}
@ -84,7 +85,8 @@ func (c *controller) Update(policy *model.Policy) error {
}
// schedule again if needed
if isScheduledTrigger(policy) {
if _, err = c.scheduler.Schedule(orm.Context(), job.Replication, policy.ID, "", policy.Trigger.Settings.Cron, CallbackFuncName, policy.ID); err != nil {
extras := make(map[string]interface{})
if _, err = c.scheduler.Schedule(orm.Context(), job.Replication, policy.ID, "", policy.Trigger.Settings.Cron, CallbackFuncName, policy.ID, extras); err != nil {
return fmt.Errorf("failed to schedule the policy %d: %v", policy.ID, err)
}
}

4
src/replication/policy/controller/controller_test.go
View File

@ -226,7 +226,7 @@ func TestCreate(t *testing.T) {
// scheduled trigger
scheduler.On("Schedule", mock.Anything, mock.Anything,
mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
_, err = ctl.Create(&model.Policy{
Enabled: true,
Trigger: &model.Trigger{
@ -269,7 +269,7 @@ func TestUpdate(t *testing.T) {
// the trigger changed
scheduler.On("Schedule", mock.Anything, mock.Anything,
mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
scheduler.On("UnScheduleByVendor", mock.Anything, mock.Anything,
mock.Anything).Return(nil)

9
src/server/v2.0/handler/model/gc.go
View File

@ -1,7 +1,9 @@
package model
import (
"encoding/json"
"github.com/go-openapi/strfmt"
"github.com/goharbor/harbor/src/lib/log"
"github.com/goharbor/harbor/src/pkg/scheduler"
"github.com/goharbor/harbor/src/server/v2.0/models"
"time"
@ -55,11 +57,16 @@ type Schedule struct {
// ToSwagger converts the schedule to the swagger model
// TODO remove the hard code when after issue https://github.com/goharbor/harbor/issues/13047 is resolved.
func (s *Schedule) ToSwagger() *models.GCHistory {
e, err := json.Marshal(s.ExtraAttrs)
if err != nil {
log.Error(err)
}
return &models.GCHistory{
ID: 0,
JobName: "",
JobKind: s.CRON,
JobParameters: s.Param,
JobParameters: string(e),
Deleted: false,
JobStatus: "",
Schedule: &models.ScheduleObj{

14
src/testing/pkg/scheduler/scheduler.go
View File

@ -62,20 +62,20 @@ func (_m *Scheduler) ListSchedules(ctx context.Context, query *q.Query) ([]*sche
return r0, r1
}
// Schedule provides a mock function with given fields: ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params
func (_m *Scheduler) Schedule(ctx context.Context, vendorType string, vendorID int64, cronType string, cron string, callbackFuncName string, params interface{}) (int64, error) {
ret := _m.Called(ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params)
// Schedule provides a mock function with given fields: ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params, extras
func (_m *Scheduler) Schedule(ctx context.Context, vendorType string, vendorID int64, cronType string, cron string, callbackFuncName string, params interface{}, extras map[string]interface{}) (int64, error) {
ret := _m.Called(ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params, extras)
var r0 int64
if rf, ok := ret.Get(0).(func(context.Context, string, int64, string, string, string, interface{}) int64); ok {
r0 = rf(ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params)
if rf, ok := ret.Get(0).(func(context.Context, string, int64, string, string, string, interface{}, map[string]interface{}) int64); ok {
r0 = rf(ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params, extras)
} else {
r0 = ret.Get(0).(int64)
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, string, int64, string, string, string, interface{}) error); ok {
r1 = rf(ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params)
if rf, ok := ret.Get(1).(func(context.Context, string, int64, string, string, string, interface{}, map[string]interface{}) error); ok {
r1 = rf(ctx, vendorType, vendorID, cronType, cron, callbackFuncName, params, extras)
} else {
r1 = ret.Error(1)
}