diff --git a/internal/sql_workbench/service/sql_workbench_service.go b/internal/sql_workbench/service/sql_workbench_service.go index ef8fb3ac..8b100566 100644 --- a/internal/sql_workbench/service/sql_workbench_service.go +++ b/internal/sql_workbench/service/sql_workbench_service.go @@ -104,6 +104,7 @@ type SqlWorkbenchService struct { sqlWorkbenchDatasourceRepo biz.SqlWorkbenchDatasourceRepo proxyTargetRepo biz.ProxyTargetRepo cbOperationLogUsecase *biz.CbOperationLogUsecase + maintenanceTimeUsecase *biz.MaintenanceTimeUsecase sqlResultMasker sqlresultmasker.SQLResultMasker } @@ -184,6 +185,7 @@ func NewAndInitSqlWorkbenchService(logger utilLog.Logger, opts *conf.DMSOptions) // 初始化操作日志相关 cbOperationLogRepo := storage.NewCbOperationLogRepo(logger, st) cbOperationLogUsecase := biz.NewCbOperationLogUsecase(logger, cbOperationLogRepo, opPermissionVerifyUsecase, proxyTargetRepo, biz.NewSystemVariableUsecase(logger, storage.NewSystemVariableRepo(logger, st))) + maintenanceTimeUsecase := biz.NewMaintenanceTimeUsecase(logger, opPermissionVerifyUsecase) return &SqlWorkbenchService{ cfg: opts.SqlWorkBenchOpts, @@ -197,6 +199,7 @@ func NewAndInitSqlWorkbenchService(logger utilLog.Logger, opts *conf.DMSOptions) sqlWorkbenchDatasourceRepo: sqlWorkbenchDatasourceRepo, proxyTargetRepo: proxyTargetRepo, cbOperationLogUsecase: cbOperationLogUsecase, + maintenanceTimeUsecase: maintenanceTimeUsecase, }, nil } @@ -1183,7 +1186,7 @@ func (sqlWorkbenchService *SqlWorkbenchService) AuditMiddleware() echo.Middlewar // 注意：解析仅服务于审核辅助路径，解析失败不应直接阻塞用户的 SQL 执行； // 否则一旦中间件辅助能力出错（如 sid 解码失败），用户连查询都跑不了。 // 真正的「未启用审核 / 审核失败」等强策略仍由后续分支按既有 fail-closed 处理。 - sql, sidInfo, err := sqlWorkbenchService.parseStreamExecuteRequest(bodyBytes) + sql, sidInfo, isExecuteAnyway, err := sqlWorkbenchService.parseStreamExecuteRequest(bodyBytes) if err != nil { sqlWorkbenchService.log.Warnf("failed to parse streamExecute request, skipping audit: %v", err) return next(c) @@ -1245,7 +1248,13 @@ func (sqlWorkbenchService *SqlWorkbenchService) AuditMiddleware() echo.Middlewar } // 拦截响应并添加审核结果 - return sqlWorkbenchService.interceptAndAddAuditResult(c, next, dmsUserId, auditResult, dbService) + execCtx := &streamExecuteRequestContext{ + sql: sql, + datasourceID: datasourceID, + schemaName: schemaName, + isExecuteAnyway: isExecuteAnyway, + } + return sqlWorkbenchService.interceptAndAddAuditResult(c, next, dmsUserId, auditResult, dbService, execCtx) } } } @@ -1257,11 +1266,11 @@ type streamExecuteSidInfo struct { dbID int } -// parseStreamExecuteRequest 解析 streamExecute 请求体，提取 SQL 与会话 sid 信息 -func (sqlWorkbenchService *SqlWorkbenchService) parseStreamExecuteRequest(bodyBytes []byte) (sql string, sidInfo *streamExecuteSidInfo, err error) { +// parseStreamExecuteRequest 解析 streamExecute 请求体，提取 SQL、会话 sid 与 isExecuteAnyway +func (sqlWorkbenchService *SqlWorkbenchService) parseStreamExecuteRequest(bodyBytes []byte) (sql string, sidInfo *streamExecuteSidInfo, isExecuteAnyway bool, err error) { var requestBody map[string]interface{} if err := json.Unmarshal(bodyBytes, &requestBody); err != nil { - return "", nil, fmt.Errorf("failed to unmarshal request body: %v", err) + return "", nil, false, fmt.Errorf("failed to unmarshal request body: %v", err) } if sqlVal, ok := requestBody["sql"]; ok { @@ -1270,6 +1279,8 @@ func (sqlWorkbenchService *SqlWorkbenchService) parseStreamExecuteRequest(bodyBy } } + isExecuteAnyway = parseIsExecuteAnywayFromRequest(requestBody) + if sidVal, ok := requestBody["sid"]; ok { if sidStr, ok := sidVal.(string); ok { sidInfo, err = sqlWorkbenchService.parseStreamExecuteSid(sidStr) @@ -1279,7 +1290,26 @@ func (sqlWorkbenchService *SqlWorkbenchService) parseStreamExecuteRequest(bodyBy } } - return sql, sidInfo, nil + return sql, sidInfo, isExecuteAnyway, nil +} + +func parseIsExecuteAnywayFromRequest(requestBody map[string]interface{}) bool { + isExecuteAnywayVal, ok := requestBody["isExecuteAnyway"] + if !ok || isExecuteAnywayVal == nil { + return false + } + + switch val := isExecuteAnywayVal.(type) { + case bool: + return val + case string: + parsed, err := strconv.ParseBool(val) + if err == nil { + return parsed + } + } + + return false } // parseStreamExecuteSid 解析 ODC sid（与 pathUtil.generateDatabaseSid 对齐）： @@ -1441,6 +1471,19 @@ func (sqlWorkbenchService *SqlWorkbenchService) isEnableSQLAudit(dbService *biz. return dbService.SQLEConfig.AuditEnabled && dbService.SQLEConfig.SQLQueryConfig.AuditEnabled } +// normalizeSQLEAuditSchemaName 将 ODC SQL Server 的 database.schema 组合归一化为 SQLE 连接库名。 +// ODC 将 SQL Server 的 database.schema 平铺为 schema 名（如 TestDB.dbo）， +// 而 SQLE 直连审核会把 schema_name 当作连接的数据库名。 +func normalizeSQLEAuditSchemaName(dbType, schemaName string) string { + if dbType != string(pkgConst.DBTypeSQLServer) { + return schemaName + } + if idx := strings.Index(schemaName, "."); idx > 0 { + return schemaName[:idx] + } + return schemaName +} + // callSQLEAudit 调用 SQLE 直接审核接口 func (sqlWorkbenchService *SqlWorkbenchService) callSQLEAudit(ctx context.Context, sql string, dbService *biz.DBService, schemaName string) (*cloudbeaver.AuditSQLReply, error) { // 获取 SQLE 服务地址 @@ -1450,6 +1493,7 @@ func (sqlWorkbenchService *SqlWorkbenchService) callSQLEAudit(ctx context.Contex } sqleAddr := fmt.Sprintf("%s/v2/sql_audit", target.URL.String()) + schemaName = normalizeSQLEAuditSchemaName(dbService.DBType, schemaName) auditReq := cloudbeaver.AuditSQLReq{ InstanceType: dbService.DBType, @@ -1480,17 +1524,22 @@ func (sqlWorkbenchService *SqlWorkbenchService) callSQLEAudit(ctx context.Contex } // interceptAndAddAuditResult 拦截响应并添加审核结果 -func (sqlWorkbenchService *SqlWorkbenchService) interceptAndAddAuditResult(c echo.Context, next echo.HandlerFunc, userId string, auditResult *cloudbeaver.AuditSQLReply, dbService *biz.DBService) error { - // 判断是否需要审批 +func (sqlWorkbenchService *SqlWorkbenchService) interceptAndAddAuditResult(c echo.Context, next echo.HandlerFunc, userId string, auditResult *cloudbeaver.AuditSQLReply, dbService *biz.DBService, execCtx *streamExecuteRequestContext) error { allowQueryWhenLessThanAuditLevel := dbService.GetAllowQueryWhenLessThanAuditLevel() - needApproval := sqlWorkbenchService.shouldRequireApproval(auditResult.Data.SQLResults, allowQueryWhenLessThanAuditLevel) + auditPassed := isAuditSuccessful(auditResult, allowQueryWhenLessThanAuditLevel) - // 如果需要审批，直接返回审核结果，不请求真实的 streamExecute 接口 - if needApproval { + if !auditPassed && !execCtx.isExecuteAnyway { return sqlWorkbenchService.buildAuditResponseWithoutExecution(c, userId, auditResult, dbService) } - // 不需要审批，执行真实请求并添加审核结果 + if blocked, err := sqlWorkbenchService.checkMaintenanceTime(c, userId, auditResult.Data.SQLResults, dbService); blocked || err != nil { + return err + } + + if sqlWorkbenchService.shouldExecuteByWorkflow(dbService, auditResult.Data.SQLResults) { + return sqlWorkbenchService.executeNonDQLByWorkflow(c, userId, execCtx, auditResult, dbService) + } + return sqlWorkbenchService.executeAndAddAuditResult(c, next, auditResult, dbService) } @@ -1799,35 +1848,68 @@ func (sqlWorkbenchService *SqlWorkbenchService) mergeSQLEAuditResults(data *Stre } } -// shouldRequireApproval 根据审核放行等级判断是否需要审批 -func (sqlWorkbenchService *SqlWorkbenchService) shouldRequireApproval(sqlResults []cloudbeaver.AuditSQLResV2, allowQueryWhenLessThanAuditLevel string) bool { - // 如果没有设置审核放行等级，那么直接放行 - if allowQueryWhenLessThanAuditLevel == "" { +// isAuditSuccessful 判断审核是否放行，语义对齐 CloudBeaver AuditSQL / IsSuccess +func isAuditSuccessful(auditResult *cloudbeaver.AuditSQLReply, allowQueryWhenLessThanAuditLevel string) bool { + if auditResult == nil || auditResult.Data == nil { return false } - // 遍历所有 SQL 审核结果 - for _, sqlResult := range sqlResults { - // 检查是否有执行失败的审核项 - for _, auditItem := range sqlResult.AuditResult { - if auditItem.ExecutionFailed { - return true + for _, sqlResult := range auditResult.Data.SQLResults { + for _, res := range sqlResult.AuditResult { + if res.ExecutionFailed { + return false } } + } - // 比较审核等级：如果 SQL 的审核等级大于允许的等级，则需要审批 - // 使用 RuleLevel 的 LessOrEqual 方法进行比较 - sqlAuditLevel := dbmodel.RuleLevel(sqlResult.AuditLevel) - allowedLevel := dbmodel.RuleLevel(allowQueryWhenLessThanAuditLevel) + if auditResult.Data.PassRate == 0 { + return cloudbeaver.AllowQuery(allowQueryWhenLessThanAuditLevel, auditResult.Data.SQLResults) + } - // 如果 SQL 的审核等级大于允许的等级（即 !LessOrEqual），则需要审批 - if !sqlAuditLevel.LessOrEqual(allowedLevel) { - return true - } + return true +} + +// shouldRequireApproval 根据审核放行语义判断是否需要审批 +func (sqlWorkbenchService *SqlWorkbenchService) shouldRequireApproval(auditResult *cloudbeaver.AuditSQLReply, allowQueryWhenLessThanAuditLevel string) bool { + return !isAuditSuccessful(auditResult, allowQueryWhenLessThanAuditLevel) +} + +// checkMaintenanceTime 检查运维时间管控（ODC 工作台） +// 返回 blocked=true 表示已构造拦截响应，调用方应立即返回 +func (sqlWorkbenchService *SqlWorkbenchService) checkMaintenanceTime(c echo.Context, userID string, auditResults []cloudbeaver.AuditSQLResV2, dbService *biz.DBService) (blocked bool, err error) { + if sqlWorkbenchService.maintenanceTimeUsecase == nil { + return false, fmt.Errorf("maintenance time usecase is nil") } - // 所有 SQL 的审核等级都小于等于允许的等级，不需要审批 - return false + if userID == "" { + return false, fmt.Errorf("current user uid is empty") + } + + sqlTypes := make([]string, 0, len(auditResults)) + for _, r := range auditResults { + sqlTypes = append(sqlTypes, r.SQLType) + } + + var sqlQueryConfig *biz.SQLQueryConfig + if dbService != nil && dbService.SQLEConfig != nil { + sqlQueryConfig = dbService.SQLEConfig.SQLQueryConfig + } + + allowed, message, checkErr := sqlWorkbenchService.maintenanceTimeUsecase.CheckSQLExecutionAllowed( + c.Request().Context(), + userID, + sqlTypes, + time.Now(), + sqlQueryConfig, + ) + if checkErr != nil { + sqlWorkbenchService.log.Errorf("check maintenance time failed: %v", checkErr) + return false, checkErr + } + if !allowed { + return true, sqlWorkbenchService.buildWorkflowErrorResponse(c, message) + } + return false, nil } // convertSQLEAuditToViolatedRules 将 SQLE 审核结果转换为 violatedRules 格式 @@ -1896,12 +1978,14 @@ type StreamExecuteResponse struct { // StreamExecuteData streamExecute 响应中的 data 字段 type StreamExecuteData struct { - ApprovalRequired bool `json:"approvalRequired"` - LogicalSQL bool `json:"logicalSql"` - RequestID *string `json:"requestId"` - SQLs []StreamExecuteSQLItem `json:"sqls"` - UnauthorizedDBResources interface{} `json:"unauthorizedDBResources"` - ViolatedRules []interface{} `json:"violatedRules"` + ApprovalRequired bool `json:"approvalRequired"` + LogicalSQL bool `json:"logicalSql"` + RequestID *string `json:"requestId"` + SQLs []StreamExecuteSQLItem `json:"sqls"` + UnauthorizedDBResources interface{} `json:"unauthorizedDBResources"` + ViolatedRules []interface{} `json:"violatedRules"` + WorkflowInfo *StreamExecuteWorkflowInfo `json:"workflowInfo,omitempty"` + ErrorMessage string `json:"errorMessage,omitempty"` } // StreamExecuteSQLItem SQL 条目 diff --git a/internal/sql_workbench/service/sql_workbench_service_test.go b/internal/sql_workbench/service/sql_workbench_service_test.go index 090ad66c..6cd3d4de 100644 --- a/internal/sql_workbench/service/sql_workbench_service_test.go +++ b/internal/sql_workbench/service/sql_workbench_service_test.go @@ -6,11 +6,129 @@ import ( "testing" "github.com/actiontech/dms/internal/dms/biz" - "github.com/actiontech/dms/internal/sql_workbench/client" pkgConst "github.com/actiontech/dms/internal/dms/pkg/constant" + dbmodel "github.com/actiontech/dms/internal/dms/storage/model" + "github.com/actiontech/dms/internal/pkg/cloudbeaver" + "github.com/actiontech/dms/internal/sql_workbench/client" pkgParams "github.com/actiontech/dms/pkg/params" ) +func Test_isAuditSuccessful(t *testing.T) { + allowWarn := string(dbmodel.RuleLevelWarn) + + cases := map[string]struct { + auditResult *cloudbeaver.AuditSQLReply + allowLevel string + expected bool + }{ + "pass rate positive": { + auditResult: &cloudbeaver.AuditSQLReply{ + Data: &cloudbeaver.AuditResDataV2{ + PassRate: 1, + SQLResults: []cloudbeaver.AuditSQLResV2{ + {AuditLevel: string(dbmodel.RuleLevelError)}, + }, + }, + }, + allowLevel: allowWarn, + expected: true, + }, + "execution failed blocks": { + auditResult: &cloudbeaver.AuditSQLReply{ + Data: &cloudbeaver.AuditResDataV2{ + PassRate: 1, + SQLResults: []cloudbeaver.AuditSQLResV2{ + { + AuditLevel: string(dbmodel.RuleLevelNormal), + AuditResult: dbmodel.AuditResults{ + {ExecutionFailed: true}, + }, + }, + }, + }, + }, + allowLevel: allowWarn, + expected: false, + }, + "pass rate zero and level within threshold": { + auditResult: &cloudbeaver.AuditSQLReply{ + Data: &cloudbeaver.AuditResDataV2{ + PassRate: 0, + SQLResults: []cloudbeaver.AuditSQLResV2{ + {AuditLevel: string(dbmodel.RuleLevelWarn)}, + }, + }, + }, + allowLevel: allowWarn, + expected: true, + }, + "pass rate zero and level above threshold": { + auditResult: &cloudbeaver.AuditSQLReply{ + Data: &cloudbeaver.AuditResDataV2{ + PassRate: 0, + SQLResults: []cloudbeaver.AuditSQLResV2{ + {AuditLevel: string(dbmodel.RuleLevelError)}, + }, + }, + }, + allowLevel: allowWarn, + expected: false, + }, + "nil audit result": { + auditResult: nil, + allowLevel: allowWarn, + expected: false, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + got := isAuditSuccessful(tc.auditResult, tc.allowLevel) + if got != tc.expected { + t.Errorf("isAuditSuccessful() = %v, want %v", got, tc.expected) + } + }) + } +} + +func Test_shouldRequireApproval(t *testing.T) { + svc := &SqlWorkbenchService{} + auditResult := &cloudbeaver.AuditSQLReply{ + Data: &cloudbeaver.AuditResDataV2{ + PassRate: 0, + SQLResults: []cloudbeaver.AuditSQLResV2{ + {AuditLevel: string(dbmodel.RuleLevelError)}, + }, + }, + } + + if !svc.shouldRequireApproval(auditResult, string(dbmodel.RuleLevelWarn)) { + t.Fatal("expected approval required for error level when pass rate is zero") + } +} + +func Test_parseIsExecuteAnywayFromRequest(t *testing.T) { + cases := map[string]struct { + body map[string]interface{} + expected bool + }{ + "bool true": {body: map[string]interface{}{"isExecuteAnyway": true}, expected: true}, + "bool false": {body: map[string]interface{}{"isExecuteAnyway": false}, expected: false}, + "string true": {body: map[string]interface{}{"isExecuteAnyway": "true"}, expected: true}, + "missing field": {body: map[string]interface{}{}, expected: false}, + "invalid string": {body: map[string]interface{}{"isExecuteAnyway": "invalid"}, expected: false}, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + got := parseIsExecuteAnywayFromRequest(tc.body) + if got != tc.expected { + t.Errorf("parseIsExecuteAnywayFromRequest() = %v, want %v", got, tc.expected) + } + }) + } +} + func Test_convertDBType(t *testing.T) { svc := &SqlWorkbenchService{} cases := map[string]struct { @@ -367,3 +485,45 @@ func Test_buildOdcCreateAndUpdateRequests_setPasswordSaved(t *testing.T) { t.Fatalf("expected passwordSaved in update JSON: %s", updateJSON) } } + +func Test_normalizeSQLEAuditSchemaName(t *testing.T) { + cases := map[string]struct { + dbType string + schema string + expected string + }{ + "SQL Server database.schema": { + dbType: string(pkgConst.DBTypeSQLServer), + schema: "TestDB.dbo", + expected: "TestDB", + }, + "SQL Server catalog only": { + dbType: string(pkgConst.DBTypeSQLServer), + schema: "TestDB", + expected: "TestDB", + }, + "SQL Server non-dbo schema": { + dbType: string(pkgConst.DBTypeSQLServer), + schema: "TestDB.sales", + expected: "TestDB", + }, + "MySQL schema unchanged": { + dbType: string(pkgConst.DBTypeMySQL), + schema: "app.db", + expected: "app.db", + }, + "Oracle schema unchanged": { + dbType: string(pkgConst.DBTypeOracle), + schema: "HR", + expected: "HR", + }, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + got := normalizeSQLEAuditSchemaName(tc.dbType, tc.schema) + if got != tc.expected { + t.Fatalf("normalizeSQLEAuditSchemaName(%q, %q) = %q, want %q", tc.dbType, tc.schema, got, tc.expected) + } + }) + } +} diff --git a/internal/sql_workbench/service/workflow_exec.go b/internal/sql_workbench/service/workflow_exec.go new file mode 100644 index 00000000..5527f503 --- /dev/null +++ b/internal/sql_workbench/service/workflow_exec.go @@ -0,0 +1,326 @@ +package sql_workbench + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "mime/multipart" + "net/http" + "strings" + "time" + + "github.com/actiontech/dms/internal/dms/biz" + "github.com/actiontech/dms/internal/dms/pkg/constant" + dbmodel "github.com/actiontech/dms/internal/dms/storage/model" + "github.com/actiontech/dms/internal/pkg/cloudbeaver" + "github.com/actiontech/dms/pkg/dms-common/i18nPkg" + _const "github.com/actiontech/dms/pkg/dms-common/pkg/const" + pkgHttp "github.com/actiontech/dms/pkg/dms-common/pkg/http" + pkgRand "github.com/actiontech/dms/pkg/rand" + "github.com/labstack/echo/v4" +) + +type streamExecuteRequestContext struct { + sql string + datasourceID string + schemaName string + isExecuteAnyway bool +} + +type workflowInstanceForCreatingTask struct { + InstanceName string `json:"instance_name"` + InstanceSchema string `json:"instance_schema"` +} + +type autoCreateAndExecuteWorkflowReq struct { + Instances []*workflowInstanceForCreatingTask `json:"instances"` + ExecMode string `json:"exec_mode"` + FileOrderMethod string `json:"file_order_method"` + Sql string `json:"sql"` + Subject string `json:"workflow_subject"` + Desc string `json:"desc"` +} + +type autoCreateAndExecuteWorkflowRes struct { + Code int `json:"code"` + Message string `json:"message"` + Data struct { + WorkflowID string `json:"workflow_id"` + WorkFlowStatus string `json:"workflow_status"` + } `json:"data"` +} + +type streamExecuteWorkflowInfo struct { + WorkflowID string `json:"workflowId"` + WorkflowStatus string `json:"workflowStatus"` + ProjectName string `json:"projectName"` + ExecSuccess bool `json:"execSuccess"` +} + +// StreamExecuteWorkflowInfo is exported for response serialization in sql_workbench_service.go. +type StreamExecuteWorkflowInfo = streamExecuteWorkflowInfo + +func (sqlWorkbenchService *SqlWorkbenchService) isEnableWorkflowExec(dbService *biz.DBService) bool { + if dbService.SQLEConfig == nil || dbService.SQLEConfig.SQLQueryConfig == nil { + return false + } + return dbService.SQLEConfig.AuditEnabled && dbService.SQLEConfig.SQLQueryConfig.WorkflowExecEnabled +} + +func (sqlWorkbenchService *SqlWorkbenchService) shouldExecuteByWorkflow(dbService *biz.DBService, auditResults []cloudbeaver.AuditSQLResV2) bool { + if !sqlWorkbenchService.isEnableSQLAudit(dbService) || !sqlWorkbenchService.isEnableWorkflowExec(dbService) { + return false + } + for _, result := range auditResults { + if result.SQLType != "" && result.SQLType != "dql" { + return true + } + } + return false +} + +func (sqlWorkbenchService *SqlWorkbenchService) checkWorkflowPermission(ctx context.Context, userUID string, dbService *biz.DBService) (bool, error) { + if userUID == constant.UIDOfUserAdmin { + return true, nil + } + opPermissions, err := sqlWorkbenchService.opPermissionVerifyUsecase.GetUserOpPermissionInProject(ctx, userUID, dbService.ProjectUID) + if err != nil { + return false, fmt.Errorf("get user op permission in project err: %v", err) + } + + requiredPermissions := map[string]struct{}{ + constant.UIDOfOpPermissionCreateWorkflow: {}, + constant.UIDOfOpPermissionAuditWorkflow: {}, + constant.UIDOfOpPermissionExecuteWorkflow: {}, + } + dbServicePermissions := make(map[string]struct{}) + + for _, opPermission := range opPermissions { + if opPermission.OpRangeType == biz.OpRangeTypeProject && opPermission.OpPermissionUID == constant.UIDOfOpPermissionProjectAdmin { + return true, nil + } + if opPermission.OpRangeType == biz.OpRangeTypeDBService { + if _, isRequired := requiredPermissions[opPermission.OpPermissionUID]; isRequired { + for _, rangeUID := range opPermission.RangeUIDs { + if rangeUID == dbService.UID { + dbServicePermissions[opPermission.OpPermissionUID] = struct{}{} + if len(dbServicePermissions) == len(requiredPermissions) { + return true, nil + } + break + } + } + } + } + } + return len(dbServicePermissions) == len(requiredPermissions), nil +} + +func (sqlWorkbenchService *SqlWorkbenchService) getSQLEURL(ctx context.Context) (string, error) { + target, err := sqlWorkbenchService.proxyTargetRepo.GetProxyTargetByName(ctx, _const.SqleComponentName) + if err != nil { + return "", fmt.Errorf("get sqle proxy target failed: %v", err) + } + return target.URL.String(), nil +} + +func (sqlWorkbenchService *SqlWorkbenchService) autoCreateAndExecuteWorkflow(ctx context.Context, projectName string, dbService *biz.DBService, sql string, instanceSchema string) (*autoCreateAndExecuteWorkflowRes, error) { + sqleURL, err := sqlWorkbenchService.getSQLEURL(ctx) + if err != nil { + return nil, err + } + + project, err := sqlWorkbenchService.projectUsecase.GetProject(ctx, dbService.ProjectUID) + if err != nil { + return nil, fmt.Errorf("get project failed: %v", err) + } + if projectName == "" { + projectName = project.Name + } + + instances := []*workflowInstanceForCreatingTask{{ + InstanceName: dbService.Name, + InstanceSchema: instanceSchema, + }} + req := autoCreateAndExecuteWorkflowReq{ + Instances: instances, + ExecMode: "sqls", + FileOrderMethod: "", + Sql: sql, + Subject: fmt.Sprintf("工作台工单_%s_%s", dbService.Name, time.Now().Format("20060102150405")), + Desc: "通过工作台执行非DQL类型的SQL时，自动创建的工单", + } + + instancesJSON, err := json.Marshal(req.Instances) + if err != nil { + return nil, fmt.Errorf("marshal instances failed: %v", err) + } + + var requestBody bytes.Buffer + writer := multipart.NewWriter(&requestBody) + fields := map[string]string{ + "instances": string(instancesJSON), + "exec_mode": req.ExecMode, + "file_order_method": req.FileOrderMethod, + "sql": req.Sql, + "workflow_subject": req.Subject, + "desc": req.Desc, + } + for key, value := range fields { + if err := writer.WriteField(key, value); err != nil { + writer.Close() + return nil, fmt.Errorf("write field %s failed: %v", key, err) + } + } + if err := writer.Close(); err != nil { + return nil, fmt.Errorf("close multipart writer failed: %v", err) + } + + url := fmt.Sprintf("%s/v1/projects/%s/workflows/auto_create_and_execute", sqleURL, projectName) + headers := map[string]string{ + "Authorization": pkgHttp.DefaultDMSToken, + } + var reply autoCreateAndExecuteWorkflowRes + if err := pkgHttp.Call(ctx, http.MethodPost, url, headers, writer.FormDataContentType(), requestBody.Bytes(), &reply); err != nil { + return nil, fmt.Errorf("request sqle failed: %v", err) + } + if reply.Code != 0 { + return nil, fmt.Errorf("sqle returned error code %d: %s", reply.Code, reply.Message) + } + return &reply, nil +} + +func (sqlWorkbenchService *SqlWorkbenchService) executeNonDQLByWorkflow( + c echo.Context, + userID string, + execCtx *streamExecuteRequestContext, + auditResult *cloudbeaver.AuditSQLReply, + dbService *biz.DBService, +) error { + ctx := c.Request().Context() + + hasPermission, err := sqlWorkbenchService.checkWorkflowPermission(ctx, userID, dbService) + if err != nil { + return sqlWorkbenchService.buildWorkflowErrorResponse(c, fmt.Sprintf("check workflow permission failed: %v", err)) + } + if !hasPermission { + return sqlWorkbenchService.buildWorkflowErrorResponse(c, "用户没有数据源上的创建、审批、上线工单权限") + } + + project, err := sqlWorkbenchService.projectUsecase.GetProject(ctx, dbService.ProjectUID) + if err != nil { + return sqlWorkbenchService.buildWorkflowErrorResponse(c, fmt.Sprintf("get project failed: %v", err)) + } + + workflowRes, err := sqlWorkbenchService.autoCreateAndExecuteWorkflow(ctx, project.Name, dbService, execCtx.sql, execCtx.schemaName) + if err != nil { + sqlWorkbenchService.log.Errorf("auto create and execute workflow failed: %v", err) + return sqlWorkbenchService.buildWorkflowErrorResponse(c, fmt.Sprintf("auto create and execute workflow failed: %v", err)) + } + sqlWorkbenchService.log.Infof("auto create and execute workflow, workflow_id: %s, status: %s", + workflowRes.Data.WorkflowID, workflowRes.Data.WorkFlowStatus) + + isExecFailed := !strings.Contains(workflowRes.Data.WorkFlowStatus, "finished") + if err := sqlWorkbenchService.saveOpLogForWorkflow(c, userID, execCtx, auditResult, dbService, workflowRes.Data.WorkflowID, isExecFailed); err != nil { + sqlWorkbenchService.log.Errorf("save operation log for workflow failed: %v", err) + } + + return sqlWorkbenchService.buildWorkflowExecuteResponse(c, project.Name, workflowRes.Data.WorkflowID, workflowRes.Data.WorkFlowStatus, !isExecFailed) +} + +func (sqlWorkbenchService *SqlWorkbenchService) buildWorkflowExecuteResponse( + c echo.Context, + projectName, workflowID, workflowStatus string, + execSuccess bool, +) error { + response := StreamExecuteResponse{ + Data: StreamExecuteData{ + ApprovalRequired: false, + LogicalSQL: false, + RequestID: nil, + SQLs: []StreamExecuteSQLItem{}, + WorkflowInfo: &StreamExecuteWorkflowInfo{ + WorkflowID: workflowID, + WorkflowStatus: workflowStatus, + ProjectName: projectName, + ExecSuccess: execSuccess, + }, + }, + DurationMillis: 0, + HTTPStatus: "OK", + RequestID: fmt.Sprintf("dms-workflow-%d", time.Now().UnixNano()), + Server: "DMS", + Successful: true, + Timestamp: float64(time.Now().Unix()), + TraceID: c.Response().Header().Get("X-Trace-ID"), + } + return c.JSON(http.StatusOK, response) +} + +func (sqlWorkbenchService *SqlWorkbenchService) buildWorkflowErrorResponse(c echo.Context, message string) error { + response := StreamExecuteResponse{ + Data: StreamExecuteData{ + ApprovalRequired: false, + LogicalSQL: false, + RequestID: nil, + SQLs: []StreamExecuteSQLItem{}, + ErrorMessage: message, + }, + DurationMillis: 0, + HTTPStatus: "OK", + RequestID: fmt.Sprintf("dms-workflow-err-%d", time.Now().UnixNano()), + Server: "DMS", + Successful: false, + Timestamp: float64(time.Now().Unix()), + TraceID: c.Response().Header().Get("X-Trace-ID"), + } + return c.JSON(http.StatusOK, response) +} + +func (sqlWorkbenchService *SqlWorkbenchService) saveOpLogForWorkflow( + c echo.Context, + userID string, + execCtx *streamExecuteRequestContext, + auditResult *cloudbeaver.AuditSQLReply, + dbService *biz.DBService, + workflowID string, + isExecFailed bool, +) error { + uid, err := pkgRand.GenStrUid() + if err != nil { + return err + } + + sessionID := extractSessionID(c.Request().URL.Path) + isAuditPass := auditResult != nil && auditResult.Data != nil && auditResult.Data.PassRate == 1 + execResult := biz.CbExecOpSuccess + if isExecFailed { + execResult = biz.CbExecOpFailure + } + + var auditResults dbmodel.AuditResults + if auditResult != nil && auditResult.Data != nil { + for _, sqlResult := range auditResult.Data.SQLResults { + auditResults = append(auditResults, sqlWorkbenchService.convertToAuditResults(&sqlResult)...) + } + } + + now := time.Now() + cbOperationLog := biz.CbOperationLog{ + UID: uid, + OpPersonUID: userID, + OpTime: &now, + DBServiceUID: dbService.UID, + OpType: biz.CbOperationLogTypeSql, + I18nOpDetail: i18nPkg.ConvertStr2I18nAsDefaultLang(execCtx.sql), + OpSessionID: &sessionID, + ProjectID: dbService.ProjectUID, + OpHost: c.RealIP(), + AuditResults: auditResults, + IsAuditPass: &isAuditPass, + ExecResult: execResult, + WorkflowID: &workflowID, + } + return sqlWorkbenchService.cbOperationLogUsecase.SaveCbOperationLog(c.Request().Context(), &cbOperationLog) +}