fix: correct the operator in the webhook payload (#18906)

Fix the incorrect or meaningless operator in the webhook payload.

Fixes: #18438

Signed-off-by: chlins <chenyuzh@vmware.com>

make/migrations/postgresql/0120_2.9.0_schema.up.sql

@@ -76,3 +76,12 @@ $$
             END LOOP;
     END
 $$;
+
+/* Refactor the structure of replication schedule callback_func_param, convert the raw id to json object for extending */
+/*       callback_func_param
+    Old:         100
+    New:  {"policy_id": 100}
+*/
+UPDATE schedule SET callback_func_param = json_build_object('policy_id', callback_func_param::int)::text
+WHERE vendor_type='REPLICATION'
+AND callback_func_param NOT LIKE '%policy_id%';

src/controller/event/handler/internal/artifact.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/goharbor/harbor/src/controller/artifact"
 	"github.com/goharbor/harbor/src/controller/event"
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/controller/repository"
 	"github.com/goharbor/harbor/src/controller/tag"
 	"github.com/goharbor/harbor/src/jobservice/job"
@@ -246,6 +247,10 @@ func (a *Handler) asyncFlushPullCount(ctx context.Context) {
 
 func (a *Handler) onPush(ctx context.Context, event *event.ArtifactEvent) error {
 	go func() {
+		if event.Operator != "" {
+			ctx = context.WithValue(ctx, operator.ContextKey{}, event.Operator)
+		}
+
 		if err := autoScan(ctx, &artifact.Artifact{Artifact: *event.Artifact}, event.Tags...); err != nil {
 			log.Errorf("scan artifact %s@%s failed, error: %v", event.Artifact.RepositoryName, event.Artifact.Digest, err)
 		}

src/controller/event/handler/replication/event/event.go

@@ -27,4 +27,5 @@ const (
 type Event struct {
 	Type     string
 	Resource *model.Resource
+	Operator string
 }

src/controller/event/handler/replication/event/handler.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/controller/replication"
 	repctlmodel "github.com/goharbor/harbor/src/controller/replication/model"
 	"github.com/goharbor/harbor/src/lib/log"
@@ -51,6 +52,10 @@ func Handle(ctx context.Context, event *Event) error {
 		return nil
 	}
 
+	if event.Operator != "" {
+		ctx = context.WithValue(ctx, operator.ContextKey{}, event.Operator)
+	}
+
 	for _, policy := range policies {
 		id, err := replication.Ctl.Start(ctx, policy, event.Resource, task.ExecutionTriggerEvent)
 		if err != nil {

src/controller/event/handler/replication/replication.go

@@ -109,6 +109,7 @@ func (r *Handler) handlePushArtifact(ctx context.Context, event *event.PushArtif
 					}},
 			},
 		},
+		Operator: event.Operator,
 	}
 	return repevent.Handle(ctx, e)
 }
@@ -140,6 +141,7 @@ func (r *Handler) handleDeleteArtifact(ctx context.Context, event *event.DeleteA
 			},
 			Deleted: true,
 		},
+		Operator: event.Operator,
 	}
 	return repevent.Handle(ctx, e)
 }
@@ -180,6 +182,7 @@ func (r *Handler) handleCreateTag(ctx context.Context, event *event.CreateTagEve
 					}},
 			},
 		},
+		Operator: event.Operator,
 	}
 	return repevent.Handle(ctx, e)
 }
@@ -212,6 +215,7 @@ func (r *Handler) handleDeleteTag(ctx context.Context, event *event.DeleteTagEve
 			Deleted:     true,
 			IsDeleteTag: true,
 		},
+		Operator: event.Operator,
 	}
 	return repevent.Handle(ctx, e)
 }

src/controller/event/handler/webhook/artifact/replication.go

@@ -151,7 +151,7 @@ func constructReplicationPayload(ctx context.Context, event *event.ReplicationEv
 	payload := &model.Payload{
 		Type:     event.EventType,
 		OccurAt:  event.OccurAt.Unix(),
-		Operator: string(execution.Trigger),
+		Operator: execution.Operator,
 		EventData: &model.EventData{
 			Replication: &ctlModel.Replication{
 				HarborHostname:     hostname,

src/controller/event/handler/webhook/artifact/retention.go

@@ -130,7 +130,7 @@ func (r *RetentionHandler) constructRetentionPayload(ctx context.Context, event
 	payload := &model.Payload{
 		Type:     event.EventType,
 		OccurAt:  event.OccurAt.Unix(),
-		Operator: execution.Trigger,
+		Operator: execution.Operator,
 		EventData: &model.EventData{
 			Retention: &evtModel.Retention{
 				Total:             event.Total,

src/controller/event/handler/webhook/quota/quota.go

@@ -107,6 +107,7 @@ func constructQuotaPayload(event *event.QuotaEvent) (*notifyModel.Payload, error
 			},
 			Custom: quotaCustom,
 		},
+		Operator: event.Operator,
 	}
 
 	if event.Resource != nil {

src/controller/event/metadata/quota.go

@@ -32,8 +32,9 @@ type QuotaMetaData struct {
 	// used to define the event topic
 	Level int
 	// the msg contains the limitation and current usage of quota
-	Msg     string
+	Msg      string
-	OccurAt time.Time
+	OccurAt  time.Time
+	Operator string
 }
 
 // Resolve quota exceed into common image event
@@ -54,6 +55,7 @@ func (q *QuotaMetaData) Resolve(evt *event.Event) error {
 		OccurAt:   q.OccurAt,
 		RepoName:  q.RepoName,
 		Msg:       q.Msg,
+		Operator:  q.Operator,
 	}
 	if q.Tag != "" || q.Digest != "" {
 		data.Resource = &event2.ImgResource{

src/controller/event/metadata/scan.go

@@ -24,14 +24,11 @@ import (
 	v1 "github.com/goharbor/harbor/src/pkg/scan/rest/v1"
 )
 
-const (
-	autoTriggeredOperator = "auto"
-)
-
 // ScanImageMetaData defines meta data of image scanning event
 type ScanImageMetaData struct {
 	Artifact *v1.Artifact
 	Status   string
+	Operator string
 }
 
 // Resolve image scanning metadata into common chart event
@@ -57,7 +54,7 @@ func (si *ScanImageMetaData) Resolve(evt *event.Event) error {
 		EventType: eventType,
 		Artifact:  si.Artifact,
 		OccurAt:   time.Now(),
-		Operator:  autoTriggeredOperator,
+		Operator:  si.Operator,
 	}
 
 	evt.Topic = topic

src/controller/event/operator/operator.go

@@ -20,12 +20,23 @@ import (
 	"github.com/goharbor/harbor/src/common/security"
 )
 
+// ContextKey is the key for storing operator in the context.
+type ContextKey struct{}
+
 // FromContext return the event operator from context
 func FromContext(ctx context.Context) string {
+	var operator string
 	sc, ok := security.FromContext(ctx)
-	if !ok {
+	if ok {
-		return ""
+		operator = sc.GetUsername()
+	}
+	// retrieve from context if not found in security context
+	if operator == "" {
+		op, ok := ctx.Value(ContextKey{}).(string)
+		if ok {
+			operator = op
+		}
 	}
 
-	return sc.GetUsername()
+	return operator
 }

src/controller/event/operator/operator_test.go

@@ -0,0 +1,47 @@
+// Copyright Project Harbor 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 operator
+
+import (
+	"context"
+	"testing"
+
+	"github.com/goharbor/harbor/src/common/security"
+	testsec "github.com/goharbor/harbor/src/testing/common/security"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFromContext(t *testing.T) {
+	{
+		// no security context and operator context should return ""
+		op := FromContext(context.Background())
+		assert.Empty(t, op)
+	}
+	{
+		// return operator from security context
+		secCtx := &testsec.Context{}
+		secCtx.On("GetUsername").Return("security-context-user")
+		ctx := security.NewContext(context.Background(), secCtx)
+		op := FromContext(ctx)
+		assert.Equal(t, "security-context-user", op)
+	}
+	{
+		// return operator from operator context
+		ctx := context.WithValue(context.Background(), ContextKey{}, "operator-context-user")
+		op := FromContext(ctx)
+		assert.Equal(t, "operator-context-user", op)
+	}
+}

src/controller/event/topic.go

@@ -307,6 +307,7 @@ type QuotaEvent struct {
 	OccurAt   time.Time
 	RepoName  string
 	Msg       string
+	Operator  string
 }
 
 func (q *QuotaEvent) String() string {

src/controller/replication/execution.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/controller/replication/flow"
 	replicationmodel "github.com/goharbor/harbor/src/controller/replication/model"
 	"github.com/goharbor/harbor/src/jobservice/job"
@@ -104,7 +105,11 @@ func (c *controller) Start(ctx context.Context, policy *replicationmodel.Policy,
 			WithMessage("the policy %d is disabled", policy.ID)
 	}
 	// create an execution record
-	id, err := c.execMgr.Create(ctx, job.ReplicationVendorType, policy.ID, trigger)
+	extra := make(map[string]interface{})
+	if op := operator.FromContext(ctx); op != "" {
+		extra["operator"] = op
+	}
+	id, err := c.execMgr.Create(ctx, job.ReplicationVendorType, policy.ID, trigger, extra)
 	if err != nil {
 		return 0, err
 	}
@@ -263,7 +268,7 @@ func (c *controller) GetTaskLog(ctx context.Context, id int64) ([]byte, error) {
 }
 
 func convertExecution(exec *task.Execution) *Execution {
-	return &Execution{
+	replicationExec := &Execution{
 		ID:            exec.ID,
 		PolicyID:      exec.VendorID,
 		Status:        exec.Status,
@@ -273,6 +278,12 @@ func convertExecution(exec *task.Execution) *Execution {
 		StartTime:     exec.StartTime,
 		EndTime:       exec.EndTime,
 	}
+
+	if operator, ok := exec.ExtraAttrs["operator"].(string); ok {
+		replicationExec.Operator = operator
+	}
+
+	return replicationExec
 }
 
 func convertTask(task *task.Task) *Task {

src/controller/replication/execution_test.go

@@ -73,7 +73,7 @@ func (r *replicationTestSuite) TestStart() {
 	r.Require().NotNil(err)
 
 	// got error when running the replication flow
-	r.execMgr.On("Create", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
+	r.execMgr.On("Create", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
 	r.execMgr.On("Get", mock.Anything, mock.Anything).Return(&task.Execution{}, nil)
 	r.execMgr.On("StopAndWait", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	r.execMgr.On("MarkError", mock.Anything, mock.Anything, mock.Anything).Return(nil)
@@ -91,7 +91,7 @@ func (r *replicationTestSuite) TestStart() {
 	r.SetupTest()
 
 	// got no error when running the replication flow
-	r.execMgr.On("Create", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
+	r.execMgr.On("Create", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil)
 	r.execMgr.On("Get", mock.Anything, mock.Anything).Return(&task.Execution{}, nil)
 	r.flowCtl.On("Start", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	r.ormCreator.On("Create").Return(nil)

src/controller/replication/model.go

@@ -28,6 +28,7 @@ type Execution struct {
 	StatusMessage string
 	Metrics       *dao.Metrics
 	Trigger       string
+	Operator      string
 	StartTime     time.Time
 	EndTime       time.Time
 }

src/controller/replication/policy.go

@@ -16,8 +16,10 @@ package replication
 
 import (
 	"context"
-	"strconv"
+	"encoding/json"
 
+	"github.com/goharbor/harbor/src/common/secret"
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/controller/replication/model"
 	"github.com/goharbor/harbor/src/jobservice/job"
 	"github.com/goharbor/harbor/src/lib/log"
@@ -31,10 +33,20 @@ const callbackFuncName = "REPLICATION_CALLBACK"
 
 func init() {
 	callbackFunc := func(ctx context.Context, param string) error {
-		policyID, err := strconv.ParseInt(param, 10, 64)
+		params := make(map[string]interface{})
-		if err != nil {
+		if err := json.Unmarshal([]byte(param), &params); err != nil {
 			return err
 		}
+
+		var policyID int64
+		if id, ok := params["policy_id"].(float64); ok {
+			policyID = int64(id)
+		}
+
+		if op, ok := params["operator"].(string); ok {
+			ctx = context.WithValue(ctx, operator.ContextKey{}, op)
+		}
+
 		policy, err := Ctl.GetPolicy(ctx, policyID)
 		if err != nil {
 			return err
@@ -121,8 +133,13 @@ func (c *controller) CreatePolicy(ctx context.Context, policy *model.Policy) (in
 	}
 	// create schedule if needed
 	if policy.IsScheduledTrigger() {
+		cbParams := map[string]interface{}{
+			"policy_id": id,
+			// the operator of schedule job is harbor-jobservice
+			"operator": secret.JobserviceUser,
+		}
 		if _, err = c.scheduler.Schedule(ctx, job.ReplicationVendorType, id, "", policy.Trigger.Settings.Cron,
-			callbackFuncName, id, map[string]interface{}{}); err != nil {
+			callbackFuncName, cbParams, map[string]interface{}{}); err != nil {
 			return 0, err
 		}
 	}
@@ -148,8 +165,13 @@ func (c *controller) UpdatePolicy(ctx context.Context, policy *model.Policy, pro
 	}
 	// create schedule if needed
 	if policy.IsScheduledTrigger() {
+		cbParams := map[string]interface{}{
+			"policy_id": policy.ID,
+			// the operator of schedule job is harbor-jobservice
+			"operator": secret.JobserviceUser,
+		}
 		if _, err := c.scheduler.Schedule(ctx, job.ReplicationVendorType, policy.ID, "", policy.Trigger.Settings.Cron,
-			callbackFuncName, policy.ID, map[string]interface{}{}); err != nil {
+			callbackFuncName, cbParams, map[string]interface{}{}); err != nil {
 			return err
 		}
 	}

src/controller/replication/policy_test.go

@@ -15,6 +15,8 @@
 package replication
 
 import (
+	"context"
+
 	repmodel "github.com/goharbor/harbor/src/controller/replication/model"
 	"github.com/goharbor/harbor/src/pkg/reg/model"
 	replicationmodel "github.com/goharbor/harbor/src/pkg/replication/model"
@@ -68,7 +70,7 @@ func (r *replicationTestSuite) TestCreatePolicy() {
 		ID: 1,
 	}, nil)
 	mock.OnAnything(r.scheduler, "Schedule").Return(int64(1), nil)
-	id, err := r.ctl.CreatePolicy(nil, &repmodel.Policy{
+	id, err := r.ctl.CreatePolicy(context.TODO(), &repmodel.Policy{
 		Name: "rule",
 		SrcRegistry: &model.Registry{
 			ID: 1,
@@ -95,7 +97,7 @@ func (r *replicationTestSuite) TestUpdatePolicy() {
 	mock.OnAnything(r.scheduler, "UnScheduleByVendor").Return(nil)
 	mock.OnAnything(r.scheduler, "Schedule").Return(int64(1), nil)
 	mock.OnAnything(r.repMgr, "Update").Return(nil)
-	err := r.ctl.UpdatePolicy(nil, &repmodel.Policy{
+	err := r.ctl.UpdatePolicy(context.TODO(), &repmodel.Policy{
 		ID:   1,
 		Name: "rule",
 		SrcRegistry: &model.Registry{

src/controller/retention/callback.go

@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/lib/log"
 	"github.com/goharbor/harbor/src/pkg/scheduler"
 )
@@ -35,6 +36,10 @@ func retentionCallback(ctx context.Context, p string) error {
 	if err := json.Unmarshal([]byte(p), param); err != nil {
 		return fmt.Errorf("failed to unmarshal the param: %v", err)
 	}
+
+	if param.Operator != "" {
+		ctx = context.WithValue(ctx, operator.ContextKey{}, param.Operator)
+	}
 	_, err := Ctl.TriggerRetentionExec(ctx, param.PolicyID, param.Trigger, false)
 	return err
 }

src/controller/retention/controller.go

@@ -19,6 +19,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/goharbor/harbor/src/common/secret"
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/jobservice/job"
 	"github.com/goharbor/harbor/src/jobservice/logger"
 	"github.com/goharbor/harbor/src/lib/q"
@@ -88,6 +90,7 @@ const (
 type TriggerParam struct {
 	PolicyID int64
 	Trigger  string
+	Operator string
 }
 
 // GetRetention Get Retention
@@ -109,6 +112,8 @@ func (r *defaultController) CreateRetention(ctx context.Context, p *policy.Metad
 			if _, err = r.scheduler.Schedule(ctx, schedulerVendorType, id, "", cron.(string), SchedulerCallback, TriggerParam{
 				PolicyID: id,
 				Trigger:  retention.ExecutionTriggerSchedule,
+				// the operator of schedule job is harbor-jobservice
+				Operator: secret.JobserviceUser,
 			}, extras); err != nil {
 				return 0, err
 			}
@@ -169,6 +174,8 @@ func (r *defaultController) UpdateRetention(ctx context.Context, p *policy.Metad
 		_, err := r.scheduler.Schedule(ctx, schedulerVendorType, p.ID, "", p.Trigger.Settings[policy.TriggerSettingsCron].(string), SchedulerCallback, TriggerParam{
 			PolicyID: p.ID,
 			Trigger:  retention.ExecutionTriggerSchedule,
+			// the operator of schedule job is harbor-jobservice
+			Operator: secret.JobserviceUser,
 		}, extras)
 		if err != nil {
 			return err
@@ -227,11 +234,11 @@ func (r *defaultController) TriggerRetentionExec(ctx context.Context, policyID i
 		return 0, err
 	}
 
-	id, err := r.execMgr.Create(ctx, job.RetentionVendorType, policyID, trigger,
+	extra := map[string]interface{}{
-		map[string]interface{}{
+		"dry_run":  dryRun,
-			"dry_run": dryRun,
+		"operator": operator.FromContext(ctx),
-		},
+	}
-	)
+	id, err := r.execMgr.Create(ctx, job.RetentionVendorType, policyID, trigger, extra)
 	if num, err := r.launcher.Launch(ctx, p, id, dryRun); err != nil {
 		if err1 := r.execMgr.StopAndWait(ctx, id, 10*time.Second); err1 != nil {
 			logger.Errorf("failed to stop the retention execution %d: %v", id, err1)
@@ -293,7 +300,7 @@ func (r *defaultController) ListRetentionExecs(ctx context.Context, policyID int
 }
 
 func convertExecution(exec *task.Execution) *retention.Execution {
-	return &retention.Execution{
+	retentionExec := &retention.Execution{
 		ID:        exec.ID,
 		PolicyID:  exec.VendorID,
 		StartTime: exec.StartTime,
@@ -303,6 +310,12 @@ func convertExecution(exec *task.Execution) *retention.Execution {
 		DryRun:    exec.ExtraAttrs["dry_run"].(bool),
 		Type:      exec.VendorType,
 	}
+
+	if operator, ok := exec.ExtraAttrs["operator"].(string); ok {
+		retentionExec.Operator = operator
+	}
+
+	return retentionExec
 }
 
 // GetTotalOfRetentionExecs Count Retention Executions

src/controller/scan/base_controller.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/goharbor/harbor/src/common/rbac"
 	ar "github.com/goharbor/harbor/src/controller/artifact"
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/controller/robot"
 	sc "github.com/goharbor/harbor/src/controller/scanner"
 	"github.com/goharbor/harbor/src/controller/tag"
@@ -308,6 +309,9 @@ func (bc *basicController) Scan(ctx context.Context, artifact *ar.Artifact, opti
 				"name": r.Name,
 			},
 		}
+		if op := operator.FromContext(ctx); op != "" {
+			extraAttrs["operator"] = op
+		}
 		executionID, err := bc.execMgr.Create(ctx, job.ImageScanJobVendorType, artifact.ID, task.ExecutionTriggerManual, extraAttrs)
 		if err != nil {
 			return err
@@ -353,7 +357,11 @@ func (bc *basicController) Stop(ctx context.Context, artifact *ar.Artifact) erro
 }
 
 func (bc *basicController) ScanAll(ctx context.Context, trigger string, async bool) (int64, error) {
-	executionID, err := bc.execMgr.Create(ctx, job.ScanAllVendorType, 0, trigger)
+	extra := make(map[string]interface{})
+	if op := operator.FromContext(ctx); op != "" {
+		extra["operator"] = op
+	}
+	executionID, err := bc.execMgr.Create(ctx, job.ScanAllVendorType, 0, trigger, extra)
 	if err != nil {
 		return 0, err
 	}
@@ -468,7 +476,18 @@ func (bc *basicController) startScanAll(ctx context.Context, executionID int64)
 		}
 	}
 
-	extraAttrs := map[string]interface{}{"summary": summary}
+	exec, err := bc.execMgr.Get(ctx, executionID)
+	if err != nil {
+		return err
+	}
+
+	extraAttrs := exec.ExtraAttrs
+	if extraAttrs == nil {
+		extraAttrs = map[string]interface{}{"summary": summary}
+	} else {
+		extraAttrs["summary"] = summary
+	}
+
 	if err := bc.execMgr.UpdateExtraAttrs(ctx, executionID, extraAttrs); err != nil {
 		log.Errorf("failed to set the summary info for the scan all execution, error: %v", err)
 		return err

src/controller/scan/base_controller_test.go

@@ -532,7 +532,8 @@ func (suite *ControllerTestSuite) TestScanAll() {
 
 		suite.execMgr.On(
 			"Create", mock.Anything, "SCAN_ALL", int64(0), "SCHEDULE",
-		).Return(executionID, nil).Once()
+			mock.Anything).Return(executionID, nil).Once()
+		suite.execMgr.On("Get", mock.Anything, mock.Anything).Return(&task.Execution{ID: executionID}, nil).Once()
 
 		mock.OnAnything(suite.accessoryMgr, "List").Return([]accessoryModel.Accessory{}, nil).Once()
 
@@ -558,7 +559,8 @@ func (suite *ControllerTestSuite) TestScanAll() {
 
 		suite.execMgr.On(
 			"Create", mock.Anything, "SCAN_ALL", int64(0), "SCHEDULE",
-		).Return(executionID, nil).Once()
+			mock.Anything).Return(executionID, nil).Once()
+		suite.execMgr.On("Get", mock.Anything, mock.Anything).Return(&task.Execution{ID: executionID}, nil).Once()
 
 		mock.OnAnything(suite.accessoryMgr, "List").Return([]accessoryModel.Accessory{}, nil).Once()
 

src/controller/scan/callback.go

@@ -16,9 +16,11 @@ package scan
 
 import (
 	"context"
+	"encoding/json"
 
 	"github.com/goharbor/harbor/src/controller/artifact"
 	"github.com/goharbor/harbor/src/controller/event/metadata"
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/controller/robot"
 	"github.com/goharbor/harbor/src/jobservice/job"
 	"github.com/goharbor/harbor/src/lib/log"
@@ -38,6 +40,7 @@ var (
 	robotCtl    = robot.Ctl
 	scanCtl     = DefaultController
 	taskMgr     = task.Mgr
+	execMgr     = task.ExecMgr
 )
 
 func init() {
@@ -56,6 +59,17 @@ func init() {
 }
 
 func scanAllCallback(ctx context.Context, param string) error {
+	if param != "" {
+		params := make(map[string]interface{})
+		if err := json.Unmarshal([]byte(param), &params); err != nil {
+			return err
+		}
+
+		if op, ok := params["operator"].(string); ok {
+			ctx = context.WithValue(ctx, operator.ContextKey{}, op)
+		}
+	}
+
 	_, err := scanCtl.ScanAll(ctx, task.ExecutionTriggerSchedule, true)
 	return err
 }
@@ -71,6 +85,11 @@ func scanTaskStatusChange(ctx context.Context, taskID int64, status string) (err
 			return err
 		}
 
+		exec, err := execMgr.Get(ctx, t.ExecutionID)
+		if err != nil {
+			return err
+		}
+
 		robotID := getRobotID(t.ExtraAttrs)
 		if robotID > 0 {
 			if err := robotCtl.Delete(ctx, robotID); err != nil {
@@ -97,6 +116,10 @@ func scanTaskStatusChange(ctx context.Context, taskID int64, status string) (err
 					},
 					Status: status,
 				}
+
+				if operator, ok := exec.ExtraAttrs["operator"].(string); ok {
+					e.Operator = operator
+				}
 				// fire event
 				notification.AddEvent(ctx, e)
 			}

src/controller/scan/callback_test.go

@@ -24,9 +24,11 @@ import (
 
 	"github.com/goharbor/harbor/src/controller/artifact"
 	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/lib/orm"
 	"github.com/goharbor/harbor/src/pkg/task"
 	artifacttesting "github.com/goharbor/harbor/src/testing/controller/artifact"
 	robottesting "github.com/goharbor/harbor/src/testing/controller/robot"
+	ormtesting "github.com/goharbor/harbor/src/testing/lib/orm"
 	"github.com/goharbor/harbor/src/testing/mock"
 	postprocessorstesting "github.com/goharbor/harbor/src/testing/pkg/scan/postprocessors"
 	reporttesting "github.com/goharbor/harbor/src/testing/pkg/scan/report"
@@ -36,6 +38,8 @@ import (
 type CallbackTestSuite struct {
 	suite.Suite
 
+	ctx context.Context
+
 	artifactCtl *artifacttesting.Controller
 
 	execMgr *tasktesting.ExecutionManager
@@ -51,6 +55,7 @@ type CallbackTestSuite struct {
 }
 
 func (suite *CallbackTestSuite) SetupSuite() {
+	suite.ctx = orm.NewContext(nil, &ormtesting.FakeOrmer{})
 	suite.artifactCtl = &artifacttesting.Controller{}
 	artifactCtl = suite.artifactCtl
 
@@ -79,56 +84,56 @@ func (suite *CallbackTestSuite) SetupSuite() {
 func (suite *CallbackTestSuite) TestScanTaskStatusChange() {
 	{
 		// get task failed
-		suite.taskMgr.On("Get", context.TODO(), int64(1)).Return(nil, fmt.Errorf("not found")).Once()
+		suite.taskMgr.On("Get", mock.Anything, int64(1)).Return(nil, fmt.Errorf("not found")).Once()
-		suite.Error(scanTaskStatusChange(context.TODO(), 1, job.SuccessStatus.String()))
+		suite.Error(scanTaskStatusChange(suite.ctx, 1, job.SuccessStatus.String()))
 	}
 
 	{
 		// status success
-		suite.taskMgr.On("Get", context.TODO(), int64(1)).Return(
+		suite.taskMgr.On("Get", mock.Anything, int64(1)).Return(
 			&task.Task{
 				ExtraAttrs: suite.makeExtraAttrs(0, 1),
 			},
 			nil,
 		).Once()
-		suite.robotCtl.On("Delete", context.TODO(), int64(1)).Return(nil).Once()
+		suite.robotCtl.On("Delete", mock.Anything, int64(1)).Return(nil).Once()
-		suite.NoError(scanTaskStatusChange(context.TODO(), 1, job.SuccessStatus.String()))
+		suite.NoError(scanTaskStatusChange(suite.ctx, 1, job.SuccessStatus.String()))
 	}
 
 	{
 		// status success, delete robot failed
-		suite.taskMgr.On("Get", context.TODO(), int64(1)).Return(
+		suite.taskMgr.On("Get", mock.Anything, int64(1)).Return(
 			&task.Task{
 				ExtraAttrs: suite.makeExtraAttrs(0, 1),
 			},
 			nil,
 		).Once()
-		suite.robotCtl.On("Delete", context.TODO(), int64(1)).Return(fmt.Errorf("failed")).Once()
+		suite.robotCtl.On("Delete", mock.Anything, int64(1)).Return(fmt.Errorf("failed")).Once()
-		suite.NoError(scanTaskStatusChange(context.TODO(), 1, job.SuccessStatus.String()))
+		suite.NoError(scanTaskStatusChange(suite.ctx, 1, job.SuccessStatus.String()))
 	}
 
 	{
 		// status success, artifact not found
-		suite.taskMgr.On("Get", context.TODO(), int64(1)).Return(
+		suite.taskMgr.On("Get", mock.Anything, int64(1)).Return(
 			&task.Task{
 				ExtraAttrs: suite.makeExtraAttrs(1, 0),
 			},
 			nil,
 		).Once()
-		suite.artifactCtl.On("Get", context.TODO(), int64(1), (*artifact.Option)(nil)).Return(nil, fmt.Errorf("not found")).Once()
+		suite.artifactCtl.On("Get", mock.Anything, int64(1), (*artifact.Option)(nil)).Return(nil, fmt.Errorf("not found")).Once()
-		suite.NoError(scanTaskStatusChange(context.TODO(), 1, job.SuccessStatus.String()))
+		suite.NoError(scanTaskStatusChange(suite.ctx, 1, job.SuccessStatus.String()))
 	}
 
 	{
 		// status success
-		suite.taskMgr.On("Get", context.TODO(), int64(1)).Return(
+		suite.taskMgr.On("Get", mock.Anything, int64(1)).Return(
 			&task.Task{
 				ExtraAttrs: suite.makeExtraAttrs(1, 0),
 			},
 			nil,
 		).Once()
-		suite.artifactCtl.On("Get", context.TODO(), int64(1), (*artifact.Option)(nil)).Return(&artifact.Artifact{}, nil).Once()
+		suite.artifactCtl.On("Get", mock.Anything, int64(1), (*artifact.Option)(nil)).Return(&artifact.Artifact{}, nil).Once()
-		suite.NoError(scanTaskStatusChange(context.TODO(), 1, job.SuccessStatus.String()))
+		suite.NoError(scanTaskStatusChange(suite.ctx, 1, job.SuccessStatus.String()))
 	}
 }
 
@@ -136,21 +141,21 @@ func (suite *CallbackTestSuite) TestScanAllCallback() {
 	{
 		// create execution failed
 		suite.execMgr.On(
-			"Create", context.TODO(), "SCAN_ALL", int64(0), "SCHEDULE",
+			"Create", mock.Anything, "SCAN_ALL", int64(0), "SCHEDULE",
-		).Return(int64(0), fmt.Errorf("failed")).Once()
+			mock.Anything).Return(int64(0), fmt.Errorf("failed")).Once()
 
-		suite.Error(scanAllCallback(context.TODO(), ""))
+		suite.Error(scanAllCallback(suite.ctx, ""))
 	}
 
 	{
 		executionID := int64(1)
 
 		suite.execMgr.On(
-			"Create", context.TODO(), "SCAN_ALL", int64(0), "SCHEDULE",
+			"Create", mock.Anything, "SCAN_ALL", int64(0), "SCHEDULE",
-		).Return(executionID, nil).Once()
+			mock.Anything).Return(executionID, nil).Once()
 
 		suite.execMgr.On(
-			"Get", context.TODO(), executionID,
+			"Get", mock.Anything, executionID,
 		).Return(&task.Execution{}, nil)
 
 		mock.OnAnything(suite.artifactCtl, "List").Return([]*artifact.Artifact{}, nil).Once()
@@ -159,7 +164,7 @@ func (suite *CallbackTestSuite) TestScanAllCallback() {
 
 		suite.execMgr.On("MarkDone", mock.Anything, executionID, mock.Anything).Return(nil).Once()
 
-		suite.NoError(scanAllCallback(context.TODO(), ""))
+		suite.NoError(scanAllCallback(suite.ctx, ""))
 	}
 }
 

src/pkg/retention/models.go

@@ -38,6 +38,7 @@ type Execution struct {
 	Status    string    `json:"status"`
 	Trigger   string    `json:"trigger"`
 	DryRun    bool      `json:"dry_run"`
+	Operator  string    `json:"operator"`
 	Type      string    `json:"-"`
 }
 

src/server/middleware/quota/util.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/goharbor/harbor/src/controller/event/metadata"
+	"github.com/goharbor/harbor/src/controller/event/operator"
 	"github.com/goharbor/harbor/src/controller/quota"
 	"github.com/goharbor/harbor/src/lib"
 	"github.com/goharbor/harbor/src/lib/log"
@@ -100,6 +101,7 @@ func projectResourcesEvent(level int) func(*http.Request, string, string, string
 			Level:    level,
 			Msg:      message,
 			OccurAt:  time.Now(),
+			Operator: operator.FromContext(ctx),
 		}
 	}
 }

src/server/v2.0/handler/scan_all.go

@@ -24,6 +24,7 @@ import (
 	"golang.org/x/text/language"
 
 	"github.com/goharbor/harbor/src/common/rbac"
+	"github.com/goharbor/harbor/src/common/secret"
 	"github.com/goharbor/harbor/src/controller/scan"
 	"github.com/goharbor/harbor/src/controller/scanner"
 	"github.com/goharbor/harbor/src/jobservice/job"
@@ -203,7 +204,11 @@ func (s *scanAllAPI) createOrUpdateScanAllSchedule(ctx context.Context, cronType
 		}
 	}
 
-	return s.scheduler.Schedule(ctx, job.ScanAllVendorType, 0, cronType, cron, scan.ScanAllCallback, nil, nil)
+	cbParams := map[string]interface{}{
+		// the operator of schedule job is harbor-jobservice
+		"operator": secret.JobserviceUser,
+	}
+	return s.scheduler.Schedule(ctx, job.ScanAllVendorType, 0, cronType, cron, scan.ScanAllCallback, cbParams, nil)
 }
 
 func (s *scanAllAPI) getScanAllSchedule(ctx context.Context) (*scheduler.Schedule, error) {

src/server/v2.0/handler/scan_all_test.go

@@ -136,8 +136,7 @@ func (suite *ScanAllTestSuite) TestAuthorization() {
 			// system admin required
 			suite.Security.On("IsAuthenticated").Return(true).Once()
 			suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(false).Once()
-			suite.Security.On("GetUsername").Return("username").Once()
+			suite.Security.On("GetUsername").Return("username")
-
 			res, err := suite.DoReq(req.method, req.url, newBody(req.body))
 			suite.NoError(err)
 			suite.Equal(403, res.StatusCode)