Webhook support (#8276)

* Webhook support: webhook manange and webhook job handle

Signed-off-by: 王添 <wangtian@corp.netease.com>
Signed-off-by: guanxiatao <guanxiatao@corp.netease.com>

make/harbor.yml

@@ -64,6 +64,10 @@ jobservice:
   # Maximum number of job workers in job service  
   max_job_workers: 10
 
+notification:
+  # Maximum retry count for webhook job
+  webhook_job_max_retry: 10
+
 chart:
   # Change the value of absolute_url to enabled can enable absolute url in chart
   absolute_url: disabled

make/migrations/postgresql/0010_1.9.0_schema.up.sql

@@ -136,3 +136,34 @@ create table schedule
 	PRIMARY KEY (id)
 );
 
+/*add notification policy table*/
+create table notification_policy (
+ id SERIAL NOT NULL,
+ name varchar(256),
+ project_id int NOT NULL,
+ enabled boolean NOT NULL DEFAULT true,
+ description text,
+ targets text,
+ event_types text,
+ creator varchar(256),
+ creation_time timestamp default CURRENT_TIMESTAMP,
+ update_time timestamp default CURRENT_TIMESTAMP,
+ PRIMARY KEY (id),
+ CONSTRAINT unique_project_id UNIQUE (project_id)
+ );
+
+/*add notification job table*/
+ CREATE TABLE notification_job (
+ id SERIAL NOT NULL,
+ policy_id int NOT NULL,
+ status varchar(32),
+ /* event_type is the type of trigger event, eg. pushImage, pullImage, uploadChart... */
+ event_type varchar(256),
+ /* notify_type is the type to notify event to user, eg. HTTP, Email... */
+ notify_type varchar(256),
+ job_detail text,
+ job_uuid varchar(64),
+ creation_time timestamp default CURRENT_TIMESTAMP,
+ update_time timestamp default CURRENT_TIMESTAMP,
+ PRIMARY KEY (id)
+ );

make/photon/prepare/templates/jobservice/env.jinja

@@ -1,3 +1,4 @@
 CORE_SECRET={{core_secret}}
 JOBSERVICE_SECRET={{jobservice_secret}}
 CORE_URL={{core_url}}
+JOBSERVICE_WEBHOOK_JOB_MAX_RETRY={{notification_webhook_job_max_retry}}

make/photon/prepare/utils/configs.py

@@ -188,6 +188,9 @@ def parse_yaml_config(config_file_path):
     config_dict['max_job_workers'] = js_config["max_job_workers"]
     config_dict['jobservice_secret'] = generate_random_string(16)
 
+    # notification config
+    notification_config = configs.get('notification') or {}
+    config_dict['notification_webhook_job_max_retry'] = notification_config["webhook_job_max_retry"]
 
     # Log configs
     allowed_levels = ['debug', 'info', 'warning', 'error', 'fatal']

src/chartserver/handler_utility.go

@@ -6,11 +6,10 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/pkg/errors"
-	"k8s.io/helm/cmd/helm/search"
-
 	hlog "github.com/goharbor/harbor/src/common/utils/log"
 	"github.com/goharbor/harbor/src/core/config"
+	"github.com/pkg/errors"
+	"k8s.io/helm/cmd/helm/search"
 )
 
 const (

src/common/config/metadata/metadatalist.go

@@ -149,6 +149,7 @@ var (
 		{Name: common.WithNotary, Scope: SystemScope, Group: BasicGroup, EnvKey: "WITH_NOTARY", DefaultValue: "false", ItemType: &BoolType{}, Editable: true},
 		// the unit of expiration is minute, 43200 minutes = 30 days
 		{Name: common.RobotTokenDuration, Scope: UserScope, Group: BasicGroup, EnvKey: "ROBOT_TOKEN_DURATION", DefaultValue: "43200", ItemType: &IntType{}, Editable: true},
+		{Name: common.NotificationEnable, Scope: UserScope, Group: BasicGroup, EnvKey: "NOTIFICATION_ENABLE", DefaultValue: "true", ItemType: &BoolType{}, Editable: true},
 
 		{Name: common.CountPerProject, Scope: UserScope, Group: QuotaGroup, EnvKey: "COUNT_PER_PROJECT", DefaultValue: "-1", ItemType: &QuotaType{}, Editable: true},
 		{Name: common.StoragePerProject, Scope: UserScope, Group: QuotaGroup, EnvKey: "STORAGE_PER_PROJECT", DefaultValue: "-1", ItemType: &QuotaType{}, Editable: true},

src/common/const.go

@@ -144,6 +144,8 @@ const (
 
 	ChartUploadCtxKey = contextKey("chart_upload_event")
 
+	// Global notification enable configuration
+	NotificationEnable = "notification_enable"
 	// Quota setting items for project
 	CountPerProject   = "count_per_project"
 	StoragePerProject = "storage_per_project"

src/common/dao/base.go

@@ -167,11 +167,13 @@ func ClearTable(table string) error {
 	return err
 }
 
-func paginateForRawSQL(sql string, limit, offset int64) string {
+// PaginateForRawSQL ...
+func PaginateForRawSQL(sql string, limit, offset int64) string {
 	return fmt.Sprintf("%s limit %d offset %d", sql, limit, offset)
 }
 
-func paginateForQuerySetter(qs orm.QuerySeter, page, size int64) orm.QuerySeter {
+// PaginateForQuerySetter ...
+func PaginateForQuerySetter(qs orm.QuerySeter, page, size int64) orm.QuerySeter {
 	if size > 0 {
 		qs = qs.Limit(size)
 		if page > 0 {

src/common/dao/notification/notification_job.go

@@ -0,0 +1,122 @@
+package notification
+
+import (
+	"fmt"
+
+	"github.com/astaxie/beego/orm"
+	"github.com/goharbor/harbor/src/common/dao"
+	"github.com/goharbor/harbor/src/common/models"
+	"github.com/goharbor/harbor/src/common/utils/log"
+	"github.com/pkg/errors"
+)
+
+// UpdateNotificationJob update notification job
+func UpdateNotificationJob(job *models.NotificationJob, props ...string) (int64, error) {
+	if job == nil {
+		return 0, errors.New("nil job")
+	}
+
+	if job.ID == 0 {
+		return 0, fmt.Errorf("notification job ID is empty")
+	}
+
+	o := dao.GetOrmer()
+	return o.Update(job, props...)
+}
+
+// AddNotificationJob insert new notification job to DB
+func AddNotificationJob(job *models.NotificationJob) (int64, error) {
+	if job == nil {
+		return 0, errors.New("nil job")
+	}
+	o := dao.GetOrmer()
+	if len(job.Status) == 0 {
+		job.Status = models.JobPending
+	}
+	return o.Insert(job)
+}
+
+// GetNotificationJob ...
+func GetNotificationJob(id int64) (*models.NotificationJob, error) {
+	o := dao.GetOrmer()
+	j := &models.NotificationJob{
+		ID: id,
+	}
+	err := o.Read(j)
+	if err == orm.ErrNoRows {
+		return nil, nil
+	}
+	return j, nil
+}
+
+// GetTotalCountOfNotificationJobs ...
+func GetTotalCountOfNotificationJobs(query ...*models.NotificationJobQuery) (int64, error) {
+	qs := notificationJobQueryConditions(query...)
+	return qs.Count()
+}
+
+// GetNotificationJobs ...
+func GetNotificationJobs(query ...*models.NotificationJobQuery) ([]*models.NotificationJob, error) {
+	var jobs []*models.NotificationJob
+
+	qs := notificationJobQueryConditions(query...)
+	if len(query) > 0 && query[0] != nil {
+		qs = dao.PaginateForQuerySetter(qs, query[0].Page, query[0].Size)
+	}
+
+	qs = qs.OrderBy("-UpdateTime")
+
+	_, err := qs.All(&jobs)
+	return jobs, err
+}
+
+// GetLastTriggerJobsGroupByEventType get notification jobs info of policy, including event type and last trigger time
+func GetLastTriggerJobsGroupByEventType(policyID int64) ([]*models.NotificationJob, error) {
+	o := dao.GetOrmer()
+	// get jobs last triggered(created) group by event_type. postgres group by usage reference:
+	// https://stackoverflow.com/questions/13325583/postgresql-max-and-group-by
+	sql := `select distinct on (event_type) event_type, id, creation_time, status, notify_type, job_uuid, update_time, 
+			creation_time, job_detail from notification_job where policy_id = ? 
+			order by event_type, id desc, creation_time, status, notify_type, job_uuid, update_time, creation_time, job_detail`
+
+	jobs := []*models.NotificationJob{}
+	_, err := o.Raw(sql, policyID).QueryRows(&jobs)
+	if err != nil {
+		log.Errorf("query last trigger info group by event type failed: %v", err)
+		return nil, err
+	}
+
+	return jobs, nil
+}
+
+// DeleteNotificationJob ...
+func DeleteNotificationJob(id int64) error {
+	o := dao.GetOrmer()
+	_, err := o.Delete(&models.NotificationJob{ID: id})
+	return err
+}
+
+// DeleteAllNotificationJobsByPolicyID ...
+func DeleteAllNotificationJobsByPolicyID(policyID int64) (int64, error) {
+	o := dao.GetOrmer()
+	return o.Delete(&models.NotificationJob{PolicyID: policyID}, "policy_id")
+}
+
+func notificationJobQueryConditions(query ...*models.NotificationJobQuery) orm.QuerySeter {
+	qs := dao.GetOrmer().QueryTable(&models.NotificationJob{})
+	if len(query) == 0 || query[0] == nil {
+		return qs
+	}
+
+	q := query[0]
+	if q.PolicyID != 0 {
+		qs = qs.Filter("PolicyID", q.PolicyID)
+	}
+	if len(q.Statuses) > 0 {
+		qs = qs.Filter("Status__in", q.Statuses)
+	}
+	if len(q.EventTypes) > 0 {
+		qs = qs.Filter("EventType__in", q.EventTypes)
+	}
+	return qs
+}