Audit plan err prompt ce

sqle/api/app.go

@@ -354,6 +354,7 @@ func StartApi(net *gracenet.Net, exitChan chan struct{}, config *config.SqleOpti
 		v1ProjectViewRouter.GET("/:project_name/sql_manages/:sql_manage_id/sql_analysis", v1.GetSqlManageSqlAnalysisV1)
 		v1ProjectViewRouter.GET("/:project_name/sql_manages/:sql_manage_id/sql_analysis_chart", v1.GetSqlManageSqlAnalysisChartV1)
 		v1ProjectViewRouter.POST("/:project_name/sql_manages/send", v1.SendSqlManage)
+		v1ProjectViewRouter.GET("/:project_name/sql_manages/abnormal_audit_plan_instance", v1.GetAbnormalInstanceAuditPlans)
 
 		// sql dev records
 		v1ProjectViewRouter.GET("/:project_name/sql_dev_records", v1.GetSqlDEVRecordList)

sqle/api/controller/v1/instance_audit_plan.go

@@ -431,10 +431,12 @@ func UpdateInstanceAuditPlanStatus(c echo.Context) error {
 }
 
 type AuditPlanTypeResBase struct {
-	AuditPlanId       uint   `json:"audit_plan_id"`
-	AuditPlanType     string `json:"type"`
-	AuditPlanTypeDesc string `json:"desc"`
-	Token             string `json:"token"`
+	AuditPlanId          uint   `json:"audit_plan_id"`
+	AuditPlanType        string `json:"type"`
+	AuditPlanTypeDesc    string `json:"desc"`
+	Token                string `json:"token"`
+	ActiveStatus         string `json:"active_status" enums:"normal,disabled"`
+	LastCollectionStatus string `json:"last_collection_status" enums:"normal,abnormal"`
 }
 
 type GetInstanceAuditPlansReqV1 struct {
@@ -580,10 +582,12 @@ func ConvertAuditPlanTypeToResByID(ctx context.Context, id string, token string)
 	for _, meta := range auditplan.Metas {
 		if meta.Type == auditPlan.Type {
 			return AuditPlanTypeResBase{
-				AuditPlanType:     auditPlan.Type,
-				AuditPlanTypeDesc: locale.Bundle.LocalizeMsgByCtx(ctx, meta.Desc),
-				AuditPlanId:       auditPlan.ID,
-				Token:             token,
+				AuditPlanType:        auditPlan.Type,
+				AuditPlanTypeDesc:    locale.Bundle.LocalizeMsgByCtx(ctx, meta.Desc),
+				AuditPlanId:          auditPlan.ID,
+				Token:                token,
+				ActiveStatus:         auditPlan.ActiveStatus,
+				LastCollectionStatus: auditPlan.AuditPlanTaskInfo.LastCollectionStatus,
 			}, nil
 		}
 	}
@@ -747,16 +751,17 @@ type AuditPlanRuleTemplate struct {
 }
 
 type InstanceAuditPlanInfo struct {
-	ID                 uint                   `json:"id"`
-	Type               AuditPlanTypeResBase   `json:"audit_plan_type"`
-	DBType             string                 `json:"audit_plan_db_type" example:"mysql"`
-	InstanceName       string                 `json:"audit_plan_instance_name" example:"test_mysql"`
-	ExecCmd            string                 `json:"exec_cmd" example:"./scanner xxx"`
-	RuleTemplate       *AuditPlanRuleTemplate `json:"audit_plan_rule_template,omitempty" `
-	TotalSQLNums       int64                  `json:"total_sql_nums"`
-	UnsolvedSQLNums    int64                  `json:"unsolved_sql_nums"`
-	LastCollectionTime *time.Time             `json:"last_collection_time"`
-	ActiveStatus       string                 `json:"active_status" enums:"normal,disabled"`
+	ID                   uint                   `json:"id"`
+	Type                 AuditPlanTypeResBase   `json:"audit_plan_type"`
+	DBType               string                 `json:"audit_plan_db_type" example:"mysql"`
+	InstanceName         string                 `json:"audit_plan_instance_name" example:"test_mysql"`
+	ExecCmd              string                 `json:"exec_cmd" example:"./scanner xxx"`
+	RuleTemplate         *AuditPlanRuleTemplate `json:"audit_plan_rule_template,omitempty" `
+	TotalSQLNums         int64                  `json:"total_sql_nums"`
+	UnsolvedSQLNums      int64                  `json:"unsolved_sql_nums"`
+	LastCollectionTime   *time.Time             `json:"last_collection_time"`
+	ActiveStatus         string                 `json:"active_status" enums:"normal,disabled"`
+	LastCollectionStatus string                 `json:"last_collection_status" enums:"normal,abnormal"`
 }
 
 // @Summary 获取实例扫描任务概览
@@ -811,15 +816,16 @@ func GetInstanceAuditPlanOverview(c echo.Context) error {
 
 		typeBase := ConvertAuditPlanTypeToRes(c.Request().Context(), v.ID, v.Type)
 		resAuditPlan := InstanceAuditPlanInfo{
-			ID:              v.ID,
-			Type:            typeBase,
-			DBType:          detail.DBType,
-			InstanceName:    inst.Name,
-			ExecCmd:         execCmd,
-			RuleTemplate:    ruleTemplate,
-			TotalSQLNums:    totalSQLNums,
-			UnsolvedSQLNums: unsolvedSQLNums,
-			ActiveStatus:    v.ActiveStatus,
+			ID:                   v.ID,
+			Type:                 typeBase,
+			DBType:               detail.DBType,
+			InstanceName:         inst.Name,
+			ExecCmd:              execCmd,
+			RuleTemplate:         ruleTemplate,
+			TotalSQLNums:         totalSQLNums,
+			UnsolvedSQLNums:      unsolvedSQLNums,
+			ActiveStatus:         v.ActiveStatus,
+			LastCollectionStatus: v.AuditPlanTaskInfo.LastCollectionStatus,
 		}
 		if v.AuditPlanTaskInfo != nil {
 			resAuditPlan.LastCollectionTime = v.AuditPlanTaskInfo.LastCollectionTime
@@ -975,6 +981,14 @@ func UpdateAuditPlanStatus(c echo.Context) error {
 		return controller.JSONBaseErrorReq(c, errors.NewAuditPlanNotExistErr())
 	}
 	auditPlan.ActiveStatus = req.Active
+	// 重启扫描任务时，重置最后采集状态
+	if req.Active == model.ActiveStatusNormal {
+		auditPlan.AuditPlanTaskInfo.LastCollectionStatus = ""
+		err = s.Save(auditPlan.AuditPlanTaskInfo)
+		if err != nil {
+			return controller.JSONBaseErrorReq(c, err)
+		}
+	}
 	err = s.Save(auditPlan)
 	if err != nil {
 		return controller.JSONBaseErrorReq(c, err)

sqle/api/controller/v1/sql_manage.go

@@ -2,6 +2,7 @@ package v1
 
 import (
 	"context"
+
 	dmsV1 "github.com/actiontech/dms/pkg/dms-common/api/dms/v1"
 	"github.com/actiontech/sqle/sqle/api/controller"
 	"github.com/actiontech/sqle/sqle/locale"
@@ -526,3 +527,26 @@ type GetGlobalSqlManageStatisticsResp struct {
 func GetGlobalSqlManageStatistics(c echo.Context) error {
 	return getGlobalSqlManageStatistics(c)
 }
+
+type GetAbnormalAuditPlanInstancesResp struct {
+	controller.BaseRes
+	Data []*AbnormalAuditPlanInstance `json:"data"`
+}
+
+type AbnormalAuditPlanInstance struct {
+	InstanceName        string `json:"instance_name" example:"MySQL"`
+	InstanceAuditPlanID uint   `json:"instance_audit_plan_id"`
+}
+
+// GetAbnormalInstanceAuditPlans get the instance of audit plan execution abnormal
+// @Summary 获取执行异常的扫描任务实例
+// @Description get the instance of audit plan execution abnormal
+// @Id getAbnormalInstanceAuditPlansV1
+// @Tags SqlManage
+// @Param project_name path string true "project name"
+// @Security ApiKeyAuth
+// @Success 200 {object} v1.GetAbnormalAuditPlanInstancesResp
+// @Router /v1/projects/{project_name}/sql_manages/abnormal_audit_plan_instance [get]
+func GetAbnormalInstanceAuditPlans(c echo.Context) error {
+	return getAbnormalInstanceAuditPlans(c)
+}

sqle/api/controller/v1/sql_manager_ce.go

@@ -51,3 +51,7 @@ func getGlobalSqlManageList(c echo.Context) error {
 func getGlobalSqlManageStatistics(c echo.Context) error {
 	return ErrCommunityEditionNotSupportSqlManage
 }
+
+func getAbnormalInstanceAuditPlans(c echo.Context) error {
+	return ErrCommunityEditionNotSupportSqlManage
+}

sqle/api/controller/v2/audit_plan.go

@@ -447,7 +447,8 @@ func PartialSyncAuditPlanSQLs(c echo.Context) error {
 }
 
 type FullSyncAuditPlanSQLsReqV2 struct {
-	SQLs []*AuditPlanSQLReqV2 `json:"audit_plan_sql_list" form:"audit_plan_sql_list" valid:"dive"`
+	SQLs         []*AuditPlanSQLReqV2 `json:"audit_plan_sql_list" form:"audit_plan_sql_list" valid:"dive"`
+	ErrorMessage string               `json:"error_message"`
 }
 
 // @Summary 全量同步SQL到扫描任务
@@ -496,7 +497,8 @@ func FullSyncAuditPlanSQLs(c echo.Context) error {
 }
 
 type UploadInstanceAuditPlanSQLsReqV2 struct {
-	SQLs []*AuditPlanSQLReqV2 `json:"audit_plan_sql_list" form:"audit_plan_sql_list" valid:"dive"`
+	ErrorMessage string               `json:"error_message"`
+	SQLs         []*AuditPlanSQLReqV2 `json:"audit_plan_sql_list" form:"audit_plan_sql_list" valid:"dive"`
 }
 
 // UploadIntanceAuditPlanSQLs
@@ -511,6 +513,7 @@ type UploadInstanceAuditPlanSQLsReqV2 struct {
 // @Success 200 {object} controller.BaseRes
 // @router /v2/projects/{project_name}/audit_plans/{audit_plan_id}/sqls/upload [post]
 func UploadInstanceAuditPlanSQLs(c echo.Context) error {
+	var err error
 	req := new(UploadInstanceAuditPlanSQLsReqV2)
 	if err := controller.BindAndValidateReq(c, req); err != nil {
 		return err
@@ -531,6 +534,12 @@ func UploadInstanceAuditPlanSQLs(c echo.Context) error {
 	}
 
 	l := log.NewEntry()
+	defer auditplan.ProcessAuditPlanStatusAndLogError(l, ap.ID, ap.InstanceID, ap.Type, &err)
+	// 当scannerd执行出现错误时，将任务状态改为异常并日志打印错误信息
+	if req.ErrorMessage != "" && len(req.SQLs) == 0 {
+		err = fmt.Errorf("error message received: %s", req.ErrorMessage)
+		return controller.JSONBaseErrorReq(c, err)
+	}
 	instance, exist, err := dms.GetInstancesById(c.Request().Context(), ap.InstanceID)
 	if err != nil {
 		return controller.JSONBaseErrorReq(c, err)
@@ -553,5 +562,9 @@ func UploadInstanceAuditPlanSQLs(c echo.Context) error {
 	if err != nil {
 		l.Errorf("update audit plan last collection time failed, error : %v", err)
 	}
-	return controller.JSONBaseErrorReq(c, auditplan.UploadSQLsV2(l, auditplan.ConvertModelToAuditPlanV2(ap), sqls))
+	err = auditplan.UploadSQLsV2(l, auditplan.ConvertModelToAuditPlanV2(ap), sqls)
+	if err != nil {
+		return controller.JSONBaseErrorReq(c, errors.NewAuditPlanExecuteExtractErr(err, ap.InstanceID, ap.Type))
+	}
+	return controller.JSONBaseErrorReq(c, nil)
 }

sqle/cmd/scannerd/scanners/common/interface_func.go

@@ -11,7 +11,7 @@ import (
 	"github.com/actiontech/sqle/sqle/pkg/scanner"
 )
 
-func Upload(ctx context.Context, sqls []scanners.SQL, c *scanner.Client, auditPlanID string) error {
+func Upload(ctx context.Context, sqls []scanners.SQL, c *scanner.Client, auditPlanID, errorMessage string) error {
 	// key=fingerPrint val=count
 	counterMap := make(map[string]uint, len(sqls))
 
@@ -34,7 +34,7 @@ func Upload(ctx context.Context, sqls []scanners.SQL, c *scanner.Client, auditPl
 		})
 	}
 
-	err := c.UploadReq(scanner.UploadSQL, auditPlanID, reqBody)
+	err := c.UploadReq(scanner.UploadSQL, auditPlanID, errorMessage, reqBody)
 	return err
 }
 

sqle/cmd/scannerd/scanners/mybatis/mybatis.go

@@ -58,6 +58,6 @@ func (mb *MyBatis) SQLs() <-chan scanners.SQL {
 	return nil
 }
 
-func (mb *MyBatis) Upload(ctx context.Context, sqls []scanners.SQL) error {
+func (mb *MyBatis) Upload(ctx context.Context, sqls []scanners.SQL, errorMessage string) error {
 	return nil
 }

sqle/cmd/scannerd/scanners/scanner.go

@@ -27,5 +27,5 @@ type Scanner interface {
 	SQLs() <-chan SQL
 
 	// Upload upload sqls to underlying client.
-	Upload(ctx context.Context, sqls []SQL) error
+	Upload(ctx context.Context, sqls []SQL, errorMessage string) error
 }

sqle/cmd/scannerd/scanners/slowquery/slowquery_ce.go

@@ -38,6 +38,6 @@ func (sq *SlowQuery) SQLs() <-chan scanners.SQL {
 	return nil
 }
 
-func (sq *SlowQuery) Upload(ctx context.Context, sqls []scanners.SQL) error {
+func (sq *SlowQuery) Upload(ctx context.Context, sqls []scanners.SQL, errorMessage string) error {
 	return errSlowQueryNotImplemented
 }

sqle/cmd/scannerd/scanners/sql_file/sqlfile.go

@@ -58,6 +58,6 @@ func (sf *SQLFile) SQLs() <-chan scanners.SQL {
 	return nil
 }
 
-func (sf *SQLFile) Upload(ctx context.Context, sqls []scanners.SQL) error {
+func (sf *SQLFile) Upload(ctx context.Context, sqls []scanners.SQL, errorMessage string) error {
 	return nil
 }

sqle/cmd/scannerd/scanners/supervisor/supervisor.go

@@ -29,6 +29,10 @@ func Start(ctx context.Context, scanner scanners.Scanner, leastPushSecond, pushB
 	for {
 		select {
 		case err := <-runErrCh:
+			UploadErr := scanner.Upload(context.TODO(), batch, err.Error())
+			if UploadErr != nil {
+				return errors.Wrap(UploadErr, "failed to upload sql")
+			}
 			return err
 
 		case <-ctx.Done():
@@ -38,7 +42,7 @@ func Start(ctx context.Context, scanner scanners.Scanner, leastPushSecond, pushB
 		case sql, ok := <-sqlCh:
 			if !ok {
 				if len(batch) != 0 {
-					err := scanner.Upload(context.TODO(), batch)
+					err := scanner.Upload(context.TODO(), batch, "")
 					if err != nil {
 						return errors.Wrap(err, "failed to upload sql")
 					}
@@ -57,7 +61,7 @@ func Start(ctx context.Context, scanner scanners.Scanner, leastPushSecond, pushB
 			}
 		}
 		logrus.StandardLogger().Infof("start uploading %d sql\n", len(batch))
-		err := scanner.Upload(context.TODO(), batch)
+		err := scanner.Upload(context.TODO(), batch, "")
 		if err != nil {
 			return errors.Wrap(err, "failed to upload sql")
 		}

sqle/cmd/scannerd/scanners/supervisor/supervisor_test.go

@@ -61,7 +61,7 @@ func (mc *mockScanner) SQLs() <-chan scanners.SQL {
 	return sqlCh
 }
 
-func (mc *mockScanner) Upload(ctx context.Context, sqls []scanners.SQL) error {
+func (mc *mockScanner) Upload(ctx context.Context, sqls []scanners.SQL, errorMessage string) error {
 	mc.uploadSQLCnt += len(sqls)
 	return nil
 }