Merge pull request #269 from reasonerjt/job-service

add tags in db, and support tags in dao and job service api

Deploy/db/registry.sql

@@ -134,6 +134,7 @@ create table replication_job (
  policy_id int NOT NULL,
  repository varchar(256) NOT NULL,
  operation  varchar(64) NOT NULL,
+ tags   varchar(16384),
  creation_time timestamp default CURRENT_TIMESTAMP,
  update_time timestamp default CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP,
  PRIMARY KEY (id)

api/jobs/replication.go

@@ -21,9 +21,10 @@ type ReplicationJob struct {
 }
 
 type ReplicationReq struct {
-	PolicyID  int64  `json:"policy_id"`
+	PolicyID  int64    `json:"policy_id"`
-	Repo      string `json:"repository"`
+	Repo      string   `json:"repository"`
-	Operation string `json:"operation"`
+	Operation string   `json:"operation"`
+	TagList   []string `json:"tags"`
 }
 
 func (rj *ReplicationJob) Post() {
@@ -64,7 +65,7 @@ func (rj *ReplicationJob) Post() {
 		} else {
 			op = models.RepOpTransfer
 		}
-		err := rj.addJob(data.Repo, data.PolicyID, op)
+		err := rj.addJob(data.Repo, data.PolicyID, op, data.TagList...)
 		if err != nil {
 			log.Errorf("Failed to insert job record, error: %v", err)
 			rj.RenderError(http.StatusInternalServerError, err.Error())
@@ -73,11 +74,12 @@ func (rj *ReplicationJob) Post() {
 	}
 }
 
-func (rj *ReplicationJob) addJob(repo string, policyID int64, operation string) error {
+func (rj *ReplicationJob) addJob(repo string, policyID int64, operation string, tags ...string) error {
 	j := models.RepJob{
 		Repository: repo,
 		PolicyID:   policyID,
 		Operation:  operation,
+		TagList:    tags,
 	}
 	log.Debugf("Creating job for repo: %s, policy: %d", repo, policyID)
 	id, err := dao.AddRepJob(j)

dao/dao_test.go

@@ -865,6 +865,7 @@ func TestAddRepJob(t *testing.T) {
 		Repository: "library/ubuntu",
 		PolicyID:   policyID,
 		Operation:  "transfer",
+		TagList:    []string{"12.01", "14.04", "latest"},
 	}
 	id, err := AddRepJob(job)
 	if err != nil {
@@ -882,9 +883,9 @@ func TestAddRepJob(t *testing.T) {
 		t.Errorf("Unable to find a job with id: %d", id)
 		return
 	}
-	if j.Status != models.JobPending || j.Repository != "library/ubuntu" || j.PolicyID != policyID || j.Operation != "transfer" {
+	if j.Status != models.JobPending || j.Repository != "library/ubuntu" || j.PolicyID != policyID || j.Operation != "transfer" || len(j.TagList) != 3 {
-		t.Errorf("Expected data of job, id: %d, Status: %s, Repository: library/ubuntu, PolicyID: %d, Operation: transfer, "+
+		t.Errorf("Expected data of job, id: %d, Status: %s, Repository: library/ubuntu, PolicyID: %d, Operation: transfer, taglist length 3"+
-			"but in returned data:, Status: %s, Repository: %s, Operation: %s, PolicyID: %d", id, models.JobPending, policyID, j.Status, j.Repository, j.Operation, j.PolicyID)
+			"but in returned data:, Status: %s, Repository: %s, Operation: %s, PolicyID: %d, TagList: %v", id, models.JobPending, policyID, j.Status, j.Repository, j.Operation, j.PolicyID, j.TagList)
 		return
 	}
 }

dao/replication_job.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/astaxie/beego/orm"
 	"github.com/vmware/harbor/models"
+	"strings"
 )
 
 func AddRepTarget(target models.RepTarget) (int64, error) {
@@ -104,8 +105,12 @@ func AddRepJob(job models.RepJob) (int64, error) {
 	if len(job.Status) == 0 {
 		job.Status = models.JobPending
 	}
+	if len(job.TagList) > 0 {
+		job.Tags = strings.Join(job.TagList, ",")
+	}
 	return o.Insert(&job)
 }
+
 func GetRepJob(id int64) (*models.RepJob, error) {
 	o := orm.NewOrm()
 	j := models.RepJob{ID: id}
@@ -113,11 +118,14 @@ func GetRepJob(id int64) (*models.RepJob, error) {
 	if err == orm.ErrNoRows {
 		return nil, nil
 	}
-	return &j, err
+	genTagListForJob(&j)
+	return &j, nil
 }
+
 func GetRepJobByPolicy(policyID int64) ([]*models.RepJob, error) {
 	var res []*models.RepJob
 	_, err := repJobPolicyIDQs(policyID).All(&res)
+	genTagListForJob(res...)
 	return res, err
 }
 
@@ -125,6 +133,7 @@ func GetRepJobByPolicy(policyID int64) ([]*models.RepJob, error) {
 func GetRepJobToStop(policyID int64) ([]*models.RepJob, error) {
 	var res []*models.RepJob
 	_, err := repJobPolicyIDQs(policyID).Filter("status__in", models.JobPending, models.JobRunning).All(&res)
+	genTagListForJob(res...)
 	return res, err
 }
 
@@ -150,3 +159,11 @@ func UpdateRepJobStatus(id int64, status string) error {
 	}
 	return err
 }
+
+func genTagListForJob(jobs ...*models.RepJob) {
+	for _, j := range jobs {
+		if len(j.Tags) > 0 {
+			j.TagList = strings.Split(j.Tags, ",")
+		}
+	}
+}

jobservice/sync_repo.json

@@ -1 +1 @@
-{"policy_id": 1, "repository":"library/ubuntu"}
+{"policy_id": 1, "repository":"library/ubuntu", "tags":["12.04","11.11"]}

models/replication_job.go

@@ -34,11 +34,13 @@ type RepPolicy struct {
 }
 
 type RepJob struct {
-	ID         int64  `orm:"column(id)" json:"id"`
+	ID         int64    `orm:"column(id)" json:"id"`
-	Status     string `orm:"column(status)" json:"status"`
+	Status     string   `orm:"column(status)" json:"status"`
-	Repository string `orm:"column(repository)" json:"repository"`
+	Repository string   `orm:"column(repository)" json:"repository"`
-	PolicyID   int64  `orm:"column(policy_id)" json:"policy_id"`
+	PolicyID   int64    `orm:"column(policy_id)" json:"policy_id"`
-	Operation  string `orm:"column(operation)" json:"operation"`
+	Operation  string   `orm:"column(operation)" json:"operation"`
+	Tags       string   `orm:"column(tags)" json:"-"`
+	TagList    []string `orm:"-" json:"tags"`
 	//	Policy       RepPolicy `orm:"-" json:"policy"`
 	CreationTime time.Time `orm:"column(creation_time);auto_now_add" json:"creation_time"`
 	UpdateTime   time.Time `orm:"column(update_time);auto_now" json:"update_time"`