Support stop purge audit log job

api/v2.0/swagger.yaml

@@ -4250,6 +4250,26 @@ paths:
           $ref: '#/responses/404'
         '500':
           $ref: '#/responses/500'
+    put:
+      summary: Stop the specific purge audit log execution
+      description: Stop the purge audit log execution specified by ID
+      tags:
+        - purge
+      operationId: stopPurge
+      parameters:
+        - $ref: '#/parameters/requestId'
+        - $ref: '#/parameters/purgeId'
+      responses:
+        '200':
+          $ref: '#/responses/200'
+        '401':
+          $ref: '#/responses/401'
+        '403':
+          $ref: '#/responses/403'
+        '404':
+          $ref: '#/responses/404'
+        '500':
+          $ref: '#/responses/500'        
   /system/purgeaudit/{purge_id}/log:
     get:
       summary: Get purge job log.

src/controller/purge/controller.go

@@ -55,13 +55,19 @@ func purgeCallback(ctx context.Context, p string) error {
 type Controller interface {
 	// Start kick off a purge schedule
 	Start(ctx context.Context, policy JobPolicy, trigger string) (int64, error)
+	// Stop a purge job
+	Stop(ctx context.Context, id int64) error
 }
 
 type controller struct {
 	taskMgr task.Manager
 	exeMgr  task.ExecutionManager
 }
 
+func (c *controller) Stop(ctx context.Context, id int64) error {
+	return c.exeMgr.Stop(ctx, id)
+}
+
 func (c *controller) Start(ctx context.Context, policy JobPolicy, trigger string) (int64, error) {
 	para := make(map[string]interface{})
 

src/jobservice/job/impl/purge/purge.go

@@ -92,7 +92,10 @@ func (j *Job) Run(ctx job.Context, params job.Parameters) error {
 	logger := ctx.GetLogger()
 	logger.Info("Purge audit job start")
 	logger.Infof("job parameters %+v", params)
-
+	if j.shouldStop(ctx) {
+		log.Info("received the stop signal, stop the purge job")
+		return nil
+	}
 	j.parseParams(params)
 	ormCtx := ctx.SystemContext()
 	if j.retentionHour == -1 || j.retentionHour == 0 {
@@ -115,3 +118,11 @@ func (j *Job) Run(ctx job.Context, params job.Parameters) error {
 	// Successfully exit
 	return nil
 }
+
+func (j *Job) shouldStop(ctx job.Context) bool {
+	opCmd, exit := ctx.OPCommand()
+	if exit && opCmd.IsStop() {
+		return true
+	}
+	return false
+}

src/jobservice/job/impl/purge/purge_test.go

@@ -63,6 +63,7 @@ func (suite *PurgeJobTestSuite) TestRun() {
 	ctx := &mockjobservice.MockJobContext{}
 	logger := &mockjobservice.MockJobLogger{}
 	ctx.On("GetLogger").Return(logger)
+	ctx.On("OPCommand").Return(job.NilCommand, true)
 	auditManager := &mockAudit.Manager{}
 	auditManager.On("Purge", mock.Anything, 128, []string{}, true).Return(int64(100), nil)
 	j := &Job{auditMgr: auditManager}
@@ -76,6 +77,15 @@ func (suite *PurgeJobTestSuite) TestRun() {
 	ret2 := j2.Run(ctx, param2)
 	suite.Require().NotNil(ret2)
 }
+func (suite *PurgeJobTestSuite) TestStop() {
+	ctx := &mockjobservice.MockJobContext{}
+	logger := &mockjobservice.MockJobLogger{}
+	ctx.On("GetLogger").Return(logger)
+	ctx.On("OPCommand").Return(job.StopCommand, true)
+	auditManager := &mockAudit.Manager{}
+	j := &Job{auditMgr: auditManager}
+	suite.True(j.shouldStop(ctx))
+}
 
 func TestPurgeJobTestSuite(t *testing.T) {
 	suite.Run(t, &PurgeJobTestSuite{})

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

@@ -303,3 +303,13 @@ func (p *purgeAPI) createSchedule(ctx context.Context, vendorType string, cronTy
 	}
 	return nil
 }
+
+func (p *purgeAPI) StopPurge(ctx context.Context, params operation.StopPurgeParams) middleware.Responder {
+	if err := p.RequireSystemAccess(ctx, rbac.ActionStop, rbac.ResourcePurgeAuditLog); err != nil {
+		return p.SendError(ctx, err)
+	}
+	if err := p.purgeCtr.Stop(ctx, params.PurgeID); err != nil {
+		return p.SendError(ctx, err)
+	}
+	return operation.NewStopPurgeOK()
+}