STRLCPY/CVE-2023-22480_KubeOperator

■ ■ ■ ■ ■ ■ ■

migration/133_add_message.up.sql

		skipped 23 lines
24	24		(UUID(), 'CLUSTER_INSTALL', 'SYSTEM', '{\"dingTalk\":\"DISABLE\",\"workWeiXin\":\"DISABLE\",\"local\":\"ENABLE\",\"email\":\"DISABLE\"}', date_add(now(), interval 8 HOUR), date_add(now(), interval 8 HOUR),'');
25	25		INSERT INTO `ko_msg_subscribe` (`id`, `name`, `type`, `config`, `created_at`, `updated_at`, `resource_id`)
26	26		VALUES
	27	+	(UUID(), 'CLUSTER_IMPORT', 'SYSTEM', '{\"dingTalk\":\"DISABLE\",\"workWeiXin\":\"DISABLE\",\"local\":\"ENABLE\",\"email\":\"DISABLE\"}', date_add(now(), interval 8 HOUR), date_add(now(), interval 8 HOUR),'');
	28	+	INSERT INTO `ko_msg_subscribe` (`id`, `name`, `type`, `config`, `created_at`, `updated_at`, `resource_id`)
	29	+	VALUES
	30	+	(UUID(), 'CLUSTER_UN_INSTALL', 'SYSTEM', '{\"dingTalk\":\"DISABLE\",\"workWeiXin\":\"DISABLE\",\"local\":\"ENABLE\",\"email\":\"DISABLE\"}', date_add(now(), interval 8 HOUR), date_add(now(), interval 8 HOUR),'');
	31	+	INSERT INTO `ko_msg_subscribe` (`id`, `name`, `type`, `config`, `created_at`, `updated_at`, `resource_id`)
	32	+	VALUES
	33	+	(UUID(), 'CLUSTER_DELETE', 'SYSTEM', '{\"dingTalk\":\"DISABLE\",\"workWeiXin\":\"DISABLE\",\"local\":\"ENABLE\",\"email\":\"DISABLE\"}', date_add(now(), interval 8 HOUR), date_add(now(), interval 8 HOUR),'');
	34	+	INSERT INTO `ko_msg_subscribe` (`id`, `name`, `type`, `config`, `created_at`, `updated_at`, `resource_id`)
	35	+	VALUES
27	36		(UUID(), 'LICENSE_EXPIRE', 'SYSTEM', '{\"dingTalk\":\"DISABLE\",\"workWeiXin\":\"DISABLE\",\"local\":\"ENABLE\",\"email\":\"DISABLE\"}', date_add(now(), interval 8 HOUR), date_add(now(), interval 8 HOUR),'');
28	37
29	38		INSERT INTO `ko_msg_subscribe` (`id`, `name`, `type`, `config`, `created_at`, `updated_at`, `resource_id`) SELECT UUID(), 'CLUSTER_OPERATOR', 'CLUSTER', '{\"dingTalk\":\"DISABLE\",\"workWeiXin\":\"DISABLE\",\"local\":\"ENABLE\",\"email\":\"DISABLE\"}', date_add(now(), interval 8 HOUR), date_add(now(), interval 8 HOUR),id from ko_cluster;
		skipped 13 lines
43	52		CREATE TABLE `ko_msg_subscribe_user` (
44	53		`subscribe_id` varchar(64) NOT NULL,
45	54		`user_id` varchar(64) NOT NULL,
46		-	`id` varchar(64) NOT NULL
	55	+	`id` varchar(64) NOT NULL,
	56	+	PRIMARY KEY (`id`)
47	57		);
48	58
49	59		INSERT INTO `ko_msg_subscribe_user` (`id`,`subscribe_id`, `user_id`)
		skipped 29 lines

■ ■ ■ ■ ■ ■

pkg/constant/message.go

		skipped 54 lines
55	55		)
56	56
57	57		var MsgTitle = map[string]string{
58		-	ClusterInstall: "集群新增",
	58	+	ClusterInstall: "集群安装",
	59	+	ClusterImport: "集群导入",
59	60		ClusterDelete: "集群删除",
60	61		ClusterUnInstall: "集群卸载",
61	62		ClusterUpgrade: "集群升级",
		skipped 87 lines

■ ■ ■ ■ ■ ■

pkg/controller/cluster_provisioner.go

		skipped 80 lines
81	81		return c.ClusterStorageProvisionerService.DeleteStorageProvisioner(clusterName, item.Name)
82	82		}
83	83
84		-	// func (c ProvisionerController) PostBatchBy(clusterName string) error {
85		-	// var batch dto.ClusterStorageProvisionerBatch
86		-	// if err := c.Ctx.ReadJSON(&batch); err != nil {
87		-	// return err
88		-	// }
89		-	// if err := c.ClusterStorageProvisionerService.BatchStorageProvisioner(clusterName, batch); err != nil {
90		-	// logger.Log.Info(fmt.Sprintf("%+v", err))
91		-	// return err
92		-	// }
93		-
94		-	// operator := c.Ctx.Values().GetString("operator")
95		-	// delClus := ""
96		-	// for _, item := range batch.Items {
97		-	// delClus += (item.Name + ",")
98		-	// }
99		-	// go kolog.Save(operator, constant.DELETE_CLUSTER_STORAGE_SUPPLIER, clusterName+"-"+delClus)
100		-
101		-	// return nil
102		-	// }
103		-

■ ■ ■ ■ ■ ■

pkg/controller/kubepi.go

		skipped 31 lines
32	32		return users, err
33	33		}
34	34
35		-	func (p KubePiController) PostBind() error {
	35	+	func (p *KubePiController) PostBind() error {
36	36		var req dto.BindKubePI
37	37		err := p.Ctx.ReadJSON(&req)
38	38		if err != nil {
		skipped 6 lines
45	45		return nil
46	46		}
47	47
48		-	func (p KubePiController) PostSearch() (*dto.BindResponse, error) {
	48	+	func (p KubePiController) PostSearch() (dto.BindResponse, error) {
49	49		var req dto.SearchBind
50	50		err := p.Ctx.ReadJSON(&req)
51	51		if err != nil {
		skipped 7 lines
59	59		return bind, nil
60	60		}
61	61
62		-	func (p KubePiController) PostCheckConn() error {
	62	+	func (p *KubePiController) PostCheckConn() error {
63	63		var req dto.CheckConn
64	64		err := p.Ctx.ReadJSON(&req)
65	65		if err != nil {
		skipped 3 lines
69	69		return p.KubePiService.CheckConn(req)
70	70		}
71	71
72		-	func (p KubePiController) GetJumpBy(project string, clusterName string) (*dto.Dashboard, error) {
	72	+	func (p KubePiController) GetJumpBy(project string, clusterName string) (dto.Dashboard, error) {
73	73		user := p.Ctx.Values().Get("user")
74	74		roleStr, _ := user.(dto.SessionUser)
75	75		userInfo, err := p.KubePiService.LoadInfo(project, clusterName, roleStr.IsAdmin)
76	76		if err != nil {
77	77		return nil, err
78		-	}
79		-	cluster, err := p.clusterRepo.GetWithPreload(clusterName, []string{"SpecConf", "Secret", "Nodes", "Nodes.Host", "Nodes.Host.Credential"})
80		-	if err != nil {
81		-	return nil, fmt.Errorf("load cluster info failed, err: %v", err.Error())
82	78		}
83	79		kubepiClient := kubepi.GetClient()
84	80		username := userInfo.Name
		skipped 6 lines
91	87		}
92	88
93	89		conn := kubepi.ConnInfo{
94		-	Name: cluster.Name,
95		-	ApiServer: fmt.Sprintf("https://%s:%d", cluster.SpecConf.LbKubeApiserverIp, cluster.SpecConf.KubeApiServerPort),
96		-	AuthenticationMode: cluster.SpecConf.AuthenticationMode,
97		-	KubernetesToken: cluster.Secret.KubernetesToken,
98		-	KeyDataStr: cluster.Secret.KeyDataStr,
99		-	CertDataStr: cluster.Secret.CertDataStr,
100		-	ConfigContent: cluster.Secret.ConfigContent,
	90	+	Name: userInfo.Cluster.Name,
	91	+	ApiServer: fmt.Sprintf("https://%s:%d", userInfo.Cluster.SpecConf.LbKubeApiserverIp, userInfo.Cluster.SpecConf.KubeApiServerPort),
	92	+	AuthenticationMode: userInfo.Cluster.SpecConf.AuthenticationMode,
	93	+	KubernetesToken: userInfo.Cluster.Secret.KubernetesToken,
	94	+	KeyDataStr: userInfo.Cluster.Secret.KeyDataStr,
	95	+	CertDataStr: userInfo.Cluster.Secret.CertDataStr,
	96	+	ConfigContent: userInfo.Cluster.Secret.ConfigContent,
101	97		}
102	98		opener, err := kubepiClient.Open(conn)
103	99		if err != nil {
		skipped 14 lines

■ ■ ■ ■ ■ ■

pkg/server/hook/mark_dirty.go

		skipped 45 lines
46	46		return err
47	47		}
48	48
	49	+	if err := tx.Model(&model.ClusterSpecComponent{}).Where("status not in (?)", []string{constant.StatusDisabled, constant.StatusEnabled, constant.StatusFailed}).Updates(map[string]interface{}{
	50	+	"status": constant.StatusFailed,
	51	+	"message": constant.TaskCancel,
	52	+	}).Error; err != nil {
	53	+	tx.Rollback()
	54	+	return err
	55	+	}
	56	+
49	57		if err := tx.Model(&model.Host{}).Where("status != ? AND status != ?", constant.StatusRunning, constant.StatusFailed).Updates(map[string]interface{}{
50	58		"status": constant.StatusFailed,
51	59		"message": constant.TaskCancel,
		skipped 33 lines

■ ■ ■ ■ ■ ■

pkg/service/cluster.go

		skipped 352 lines
353	353		return fmt.Errorf("cluster %s already in status %s", cluster.Name, cluster.Status)
354	354		}
355	355		case constant.ClusterSourceExternal:
356		-	_ = c.msgService.SendMsg(constant.ClusterDelete, constant.Cluster, cluster, true, map[string]string{})
	356	+	_ = c.msgService.SendMsg(constant.ClusterDelete, constant.System, cluster, true, map[string]string{"detailName": cluster.Name})
357	357		if err := db.DB.Delete(&cluster).Error; err != nil {
358	358		return err
359	359		}
		skipped 8 lines
368	368		_ = c.clusterRepo.Save(cluster)
369	369		_ = c.tasklogService.End(&cluster.TaskLog, false, errStr.Error())
370	370
371		-	_ = c.msgService.SendMsg(constant.ClusterUnInstall, constant.Cluster, &cluster, false, map[string]string{"errMsg": errStr.Error()})
	371	+	_ = c.msgService.SendMsg(constant.ClusterUnInstall, constant.System, cluster, false, map[string]string{"errMsg": errStr.Error(), "detailName": cluster.Name})
372	372		}
373	373
374	374		const terminalPlaybookName = "99-reset-cluster.yml"
		skipped 32 lines
407	407		}
408	408		return
409	409		}
410		-	_ = c.msgService.SendMsg(constant.ClusterUnInstall, constant.Cluster, &cluster, true, map[string]string{})
	410	+	_ = c.msgService.SendMsg(constant.ClusterUnInstall, constant.System, cluster, true, map[string]string{"detailName": cluster.Name})
411	411		if err := c.tasklogService.End(&cluster.TaskLog, true, ""); err != nil {
412	412		logger.Log.Errorf("update tasklog error %s", err.Error())
413	413		}
		skipped 25 lines
439	439		}
440	440		return
441	441		}
442		-	_ = c.msgService.SendMsg(constant.ClusterUnInstall, constant.Cluster, &cluster, true, map[string]string{})
	442	+	_ = c.msgService.SendMsg(constant.ClusterUnInstall, constant.System, cluster, true, map[string]string{"detailName": cluster.Name})
443	443		logger.Log.Infof("start clearing cluster data %s", cluster.Name)
444	444		if err := db.DB.Delete(&cluster).Error; err != nil {
445	445		c.errClusterDelete(cluster, err)
		skipped 132 lines

■ ■ ■ ■ ■ ■

pkg/service/cluster_iaas.go

		skipped 141 lines
142	142		tx.Rollback()
143	143		return err
144	144		}
	145	+	if err := tx.Save(host).Error; err != nil {
	146	+	tx.Rollback()
	147	+	return err
	148	+	}
145	149		}
146	150
147	151		nodes, err := c.createNodes(*cluster, hosts)
		skipped 403 lines

■ ■ ■ ■ ■ ■

pkg/service/cluster_import.go

		skipped 209 lines
210	210		return fmt.Errorf("can not create project resource %s", err.Error())
211	211		}
212	212		tx.Commit()
213		-	_ = c.msgService.SendMsg(constant.ClusterImport, constant.Cluster, &cluster, true, map[string]string{})
	213	+	_ = c.msgService.SendMsg(constant.ClusterImport, constant.System, cluster, true, map[string]string{"detailName": cluster.Name})
214	214
215	215		hostService := NewHostService()
216	216		go func() {
		skipped 4 lines
221	221
222	222		func (c clusterImportService) handlerImportError(tx gorm.DB, cluster model.Cluster, err error) {
223	223		tx.Rollback()
224		-	_ = c.msgService.SendMsg(constant.ClusterImport, constant.Cluster, &cluster, false, map[string]string{"errMsg": err.Error()})
	224	+	_ = c.msgService.SendMsg(constant.ClusterImport, constant.Cluster, cluster, false, map[string]string{"errMsg": err.Error(), "detailName": cluster.Name})
225	225		}
226	226
227	227		func gatherClusterInfo(cluster *model.Cluster) error {
		skipped 303 lines

■ ■ ■ ■ ■ ■

pkg/service/cluster_init.go

		skipped 55 lines
56	56		cluster.Message = err.Error()
57	57		_ = c.clusterRepo.Save(&cluster)
58	58		logger.Log.Errorf("init cluster resource for create failed: %s", err.Error())
59		-	_ = c.msgService.SendMsg(constant.ClusterInstall, constant.System, cluster, false, map[string]string{"errMsg": err.Error()})
	59	+	_ = c.msgService.SendMsg(constant.ClusterInstall, constant.System, cluster, false, map[string]string{"errMsg": err.Error(), "detailName": cluster.Name})
60	60		return
61	61		}
62	62		}
		skipped 23 lines
86	86		cluster.Message = result.Message
87	87		_ = c.clusterRepo.Save(&cluster)
88	88		logger.Log.Errorf("cluster install failed: %s", cluster.TaskLog.Message)
89		-	_ = c.msgService.SendMsg(constant.ClusterInstall, constant.System, cluster, false, map[string]string{"errMsg": cluster.TaskLog.Message})
	89	+	_ = c.msgService.SendMsg(constant.ClusterInstall, constant.System, cluster, false, map[string]string{"errMsg": cluster.TaskLog.Message, "detailName": cluster.Name})
90	90		return
91	91		case constant.TaskLogStatusSuccess:
92	92		if err := c.taskLogService.End(&cluster.TaskLog, true, ""); err != nil {
		skipped 3 lines
96	96		cluster.Status = constant.StatusRunning
97	97		cluster.Message = result.Message
98	98		cluster.CurrentTaskID = ""
99		-	_ = c.msgService.SendMsg(constant.ClusterInstall, constant.System, cluster, true, map[string]string{})
	99	+	_ = c.msgService.SendMsg(constant.ClusterInstall, constant.System, cluster, true, map[string]string{"detailName": cluster.Name})
100	100		firstMasterIP := ""
101	101		for i := range cluster.Nodes {
102	102		if cluster.Nodes[i].Role == constant.NodeRoleNameMaster && len(firstMasterIP) == 0 {
		skipped 103 lines

■ ■ ■ ■ ■ ■

pkg/service/cluster_node.go

		skipped 273 lines
274	274		return
275	275		}
276	276		c.updateNodeStatus(cluster, constant.ClusterRemoveWorker, constant.StatusRunning, nodeIDs, nil)
277		-	_ = c.msgService.SendMsg(constant.ClusterRemoveWorker, constant.Cluster, &cluster, true, map[string]string{})
	277	+	_ = c.msgService.SendMsg(constant.ClusterRemoveWorker, constant.Cluster, cluster, true, map[string]string{})
278	278		logger.Log.Info("delete node successful!")
279	279		}
280	280
		skipped 174 lines

■ ■ ■ ■ ■ ■

pkg/service/cluster_node_create.go

		skipped 339 lines
340	340		}
341	341		if status == constant.StatusFailed {
342	342		taskSuccess = false
343		-	_ = c.msgService.SendMsg(operation, constant.Cluster, &cluster, false, map[string]string{"errMsg": errmsg})
	343	+	_ = c.msgService.SendMsg(operation, constant.Cluster, cluster, false, map[string]string{"errMsg": errmsg})
	344	+	} else {
	345	+	_ = c.msgService.SendMsg(operation, constant.Cluster, cluster, true, map[string]string{})
344	346		}
345	347		_ = c.taskLogService.End(&cluster.TaskLog, taskSuccess, errmsg)
346	348
		skipped 58 lines

■ ■ ■ ■ ■ ■

pkg/service/cluster_storage_provisioner.go

		skipped 121 lines
122	122		return errors.New("load image pull port failed")
123	123		}
124	124		}
	125	+	creation.Vars["registry_port"] = fmt.Sprint(registery.RegistryPort)
125	126		playbook := c.loadPlayBookName(dp, creation.Vars)
126	127		task := model.TaskLogDetail{
127	128		ID: dp.ID,
		skipped 6 lines
134	135		return fmt.Errorf("save tasklog failed, err: %v", err)
135	136		}
136	137
137		-	admCluster, writer, err := c.loadAdmCluster(cluster, dp, creation.Vars, constant.StatusEnabled)
138		-	if err != nil {
139		-	return err
140		-	}
141		-	admCluster.Kobe.SetVar("registry_port", fmt.Sprint(registery.RegistryPort))
142	138		if err := db.DB.Model(&model.ClusterStorageProvisioner{}).Where("id = ?", dp.ID).
143	139		Updates(map[string]interface{}{"status": constant.StatusInitializing, "message": ""}).Error; err != nil {
144	140		return err
145	141		}
146		-	client, err := clusterUtil.NewClusterClient(&cluster)
147		-	if err != nil {
148		-	_ = c.taskLogService.EndDetail(&task, "provisioner", constant.TaskLogStatusFailed, err.Error())
149		-	c.errHandlerProvisioner(dp, constant.StatusDisabled, err)
150		-	}
151	142
152	143		//playbook
153		-	go c.docreate(admCluster, writer, task, dp, client)
	144	+	go c.docreate(&cluster, task, dp, creation.Vars)
154	145		return nil
155	146		}
156	147
157		-	func (c clusterStorageProvisionerService) docreate(admCluster adm.AnsibleHelper, writer io.Writer, task model.TaskLogDetail, dp model.ClusterStorageProvisioner, client kubernetes.Clientset) {
	148	+	func (c clusterStorageProvisionerService) docreate(cluster *model.Cluster, task model.TaskLogDetail, dp model.ClusterStorageProvisioner, vars map[string]interface{}) {
	149	+	admCluster, writer, err := c.loadAdmCluster(*cluster, dp, vars, constant.StatusEnabled)
	150	+	if err != nil {
	151	+	_ = c.taskLogService.EndDetail(&task, dp.Name, "provisioner", constant.TaskLogStatusFailed, err.Error())
	152	+	c.errHandlerProvisioner(dp, constant.StatusFailed, err)
	153	+	return
	154	+	}
	155	+
	156	+	client, err := clusterUtil.NewClusterClient(cluster)
	157	+	if err != nil {
	158	+	_ = c.taskLogService.EndDetail(&task, dp.Name, "provisioner", constant.TaskLogStatusFailed, err.Error())
	159	+	c.errHandlerProvisioner(dp, constant.StatusFailed, err)
	160	+	return
	161	+	}
	162	+
158	163		playbook := strings.ReplaceAll(task.Task, " (enable)", "")
159	164		if err := phases.RunPlaybookAndGetResult(admCluster.Kobe, playbook, "", writer); err != nil {
160		-	_ = c.taskLogService.EndDetail(&task, "provisioner", constant.TaskLogStatusFailed, err.Error())
	165	+	_ = c.taskLogService.EndDetail(&task, dp.Name, "provisioner", constant.TaskLogStatusFailed, err.Error())
161	166		c.errHandlerProvisioner(dp, constant.StatusFailed, err)
162	167		return
163	168		}
164		-	_ = c.taskLogService.EndDetail(&task, "provisioner", constant.TaskLogStatusSuccess, "")
	169	+	_ = c.taskLogService.EndDetail(&task, dp.Name, "provisioner", constant.TaskLogStatusSuccess, "")
165	170		dp.Status = constant.StatusWaiting
166	171		if err := db.DB.Save(&dp).Error; err != nil {
167	172		logger.Log.Errorf("save storage provisioner status err: %s", err.Error())
		skipped 35 lines
203	208		return fmt.Errorf("save tasklog failed, err: %v", err)
204	209		}
205	210
206		-	admCluster, writer, err := c.loadAdmCluster(cluster, provisioner, Vars, constant.StatusDisabled)
207		-	if err != nil {
208		-	return err
209		-	}
210	211		if err := db.DB.Model(&model.ClusterStorageProvisioner{}).Where("id = ?", provisioner.ID).
211	212		Updates(map[string]interface{}{"status": constant.StatusTerminating, "message": ""}).Error; err != nil {
212	213		return err
213	214		}
214	215
215		-	go c.dodelete(admCluster, writer, task, provisioner)
	216	+	go c.dodelete(&cluster, task, provisioner, Vars)
216	217
217	218		return nil
218	219		}
219	220
220		-	func (c clusterStorageProvisionerService) dodelete(admCluster *adm.AnsibleHelper, writer io.Writer, task model.TaskLogDetail, provisioner model.ClusterStorageProvisioner) {
	221	+	func (c clusterStorageProvisionerService) dodelete(cluster *model.Cluster, task model.TaskLogDetail, provisioner model.ClusterStorageProvisioner, vars map[string]interface{}) {
	222	+	admCluster, writer, err := c.loadAdmCluster(*cluster, provisioner, vars, constant.StatusDisabled)
	223	+	if err != nil {
	224	+	_ = c.taskLogService.EndDetail(&task, provisioner.Name, "provisioner", constant.TaskLogStatusFailed, err.Error())
	225	+	c.errHandlerProvisioner(provisioner, constant.StatusFailed, err)
	226	+	return
	227	+	}
	228	+
221	229		playbook := strings.ReplaceAll(task.Task, " (disable)", "")
222	230		if err := phases.RunPlaybookAndGetResult(admCluster.Kobe, playbook, "", writer); err != nil {
223		-	_ = c.taskLogService.EndDetail(&task, "provisioner", constant.TaskLogStatusFailed, err.Error())
	231	+	_ = c.taskLogService.EndDetail(&task, provisioner.Name, "provisioner", constant.TaskLogStatusFailed, err.Error())
224	232		c.errHandlerProvisioner(provisioner, constant.StatusFailed, err)
225	233		return
226	234		}
227		-	_ = c.taskLogService.EndDetail(&task, "provisioner", constant.TaskLogStatusSuccess, "")
	235	+	_ = c.taskLogService.EndDetail(&task, provisioner.Name, "provisioner", constant.TaskLogStatusSuccess, "")
228	236		_ = db.DB.Where("id = ?", provisioner.ID).Delete(&model.ClusterStorageProvisioner{})
229	237		}
230	238
		skipped 152 lines
383	391		if err != nil {
384	392		return admCluster, writer, err
385	393		}
386		-	if err := db.DB.Model(&model.ClusterStorageProvisioner{}).Where("id = ?", provisioner.ID).Updates(map[string]interface{}{
387		-	"status": constant.StatusTerminating,
388		-	"message": "",
389		-	}).Error; err != nil {
390		-	return admCluster, writer, err
391		-	}
392	394
393	395		switch provisioner.Type {
394	396		case "nfs":
395	397		admCluster.Kobe.SetVar(facts.EnableNfsFactName, operation)
396	398		case "gfs":
397	399		admCluster.Kobe.SetVar(facts.EnableGfsFactName, operation)
398		-	case "external_ceph_block":
	400	+	case "external-ceph-block":
	401	+	fmt.Println(facts.EnableCephBlockFactName, operation)
399	402		admCluster.Kobe.SetVar(facts.EnableCephBlockFactName, operation)
400		-	case "external_cephfs":
	403	+	case "external-cephfs":
401	404		admCluster.Kobe.SetVar(facts.EnableCephFsFactName, operation)
402	405		case "cinder":
403	406		admCluster.Kobe.SetVar(facts.EnableCinderFactName, operation)
		skipped 10 lines

■ ■ ■ ■ ■ ■

pkg/service/cluster_upgrade.go

		skipped 82 lines
83	83		// 创建日志
84	84		writer, err := ansible.CreateAnsibleLogWriterWithId(cluster.Name, cluster.TaskLog.ID)
85	85		if err != nil {
86		-	_ = c.msgService.SendMsg(constant.ClusterUpgrade, constant.Cluster, cluster, false, map[string]string{"errMsg": err.Error()})
	86	+	_ = c.msgService.SendMsg(constant.ClusterUpgrade, constant.Cluster, cluster, false, map[string]string{"errMsg": err.Error(), "detailName": cluster.Name})
87	87		return fmt.Errorf("create log error %s", err.Error())
88	88		}
89	89		if len(upgrade.Version) != 0 {
		skipped 2 lines
92	92		cluster.Status = constant.StatusUpgrading
93	93		cluster.CurrentTaskID = cluster.TaskLog.ID
94	94		if err := c.clusterRepo.Save(&cluster); err != nil {
95		-	_ = c.msgService.SendMsg(constant.ClusterUpgrade, constant.Cluster, cluster, false, map[string]string{"errMsg": err.Error()})
	95	+	_ = c.msgService.SendMsg(constant.ClusterUpgrade, constant.Cluster, cluster, false, map[string]string{"errMsg": err.Error(), "detailName": cluster.Name})
96	96		return fmt.Errorf("save cluster spec error %s", err.Error())
97	97		}
98	98		// 更新工具版本状态
		skipped 13 lines
112	112		go c.doUpgrade(ctx, *admCluster, statusChan)
113	113		for {
114	114		result := <-statusChan
115		-	// 保存进度
116		-	cluster.Status = result.Status
117		-	cluster.Message = result.Message
118		-	_ = c.clusterRepo.Save(cluster)
119	115		switch cluster.TaskLog.Phase {
120	116		case constant.TaskLogStatusSuccess:
121	117		if err := c.taskLogService.End(&cluster.TaskLog, true, ""); err != nil {
		skipped 5 lines
127	123		cluster.CurrentTaskID = ""
128	124		_ = c.clusterRepo.Save(cluster)
129	125
130		-	_ = c.msgService.SendMsg(constant.ClusterUpgrade, constant.Cluster, &cluster, true, map[string]string{})
	126	+	_ = c.msgService.SendMsg(constant.ClusterUpgrade, constant.Cluster, cluster, true, map[string]string{"detailName": cluster.Name})
131	127		cluster.Version = cluster.UpgradeVersion
132	128		_ = db.DB.Save(&cluster).Error
133	129		cancel()
		skipped 7 lines
141	137		cluster.Message = result.Message
142	138		_ = c.clusterRepo.Save(cluster)
143	139
144		-	_ = c.msgService.SendMsg(constant.ClusterUpgrade, constant.Cluster, &cluster, false, map[string]string{"errMsg": result.Message})
	140	+	_ = c.msgService.SendMsg(constant.ClusterUpgrade, constant.Cluster, cluster, false, map[string]string{"errMsg": result.Message, "detailName": cluster.Name})
145	141		cancel()
146	142		return
147	143		default:
		skipped 60 lines

■ ■ ■ ■ ■ ■

pkg/service/component.go

		skipped 152 lines
153	153		}
154	154		component = creation.ComponentCreate2Mo()
155	155		component.ClusterID = cluster.ID
	156	+	component.Status = constant.StatusInitializing
156	157		if err := db.DB.Create(&component).Error; err != nil {
157	158		return err
158	159		}
	160	+
159	161		playbook := c.loadPlayBookName(creation.Name)
160	162		task := model.TaskLogDetail{
161	163		ID: component.ID,
		skipped 3 lines
165	167		Status: constant.TaskLogStatusRunning,
166	168		}
167	169		if err := c.taskLogService.StartDetail(&task); err != nil {
	170	+	c.errHandlerComponent(component, constant.StatusDisabled, err)
168	171		return fmt.Errorf("save tasklog failed, err: %v", err)
169	172		}
170	173
171		-	admCluster, writer, err := c.loadAdmCluster(cluster, component, creation.Vars, constant.StatusEnabled)
	174	+	go c.docreate(&cluster, task, component, creation.Vars)
	175	+	return nil
	176	+	}
	177	+
	178	+	func (c componentService) docreate(cluster *model.Cluster, task model.TaskLogDetail, component model.ClusterSpecComponent, vars map[string]interface{}) {
	179	+	admCluster, writer, err := c.loadAdmCluster(*cluster, component, vars, constant.StatusEnabled)
172	180		if err != nil {
173		-	return err
	181	+	_ = c.taskLogService.EndDetail(&task, component.Name, "component", constant.TaskLogStatusFailed, err.Error())
	182	+	c.errHandlerComponent(component, constant.StatusDisabled, err)
	183	+	return
174	184		}
175	185
176		-	if err := db.DB.Model(&model.ClusterSpecComponent{}).Where("id = ?", component.ID).
177		-	Updates(map[string]interface{}{"status": constant.StatusInitializing, "message": ""}).Error; err != nil {
178		-	return err
179		-	}
180		-	client, err := clusterUtil.NewClusterClient(&cluster)
	186	+	client, err := clusterUtil.NewClusterClient(cluster)
181	187		if err != nil {
182		-	_ = c.taskLogService.EndDetail(&task, "component", constant.TaskLogStatusFailed, err.Error())
	188	+	_ = c.taskLogService.EndDetail(&task, component.Name, "component", constant.TaskLogStatusFailed, err.Error())
183	189		c.errHandlerComponent(component, constant.StatusDisabled, err)
	190	+	return
184	191		}
185	192
186		-	go c.docreate(admCluster, writer, task, component, client)
187		-	return nil
188		-	}
189		-
190		-	func (c componentService) docreate(admCluster adm.AnsibleHelper, writer io.Writer, task model.TaskLogDetail, component model.ClusterSpecComponent, client kubernetes.Clientset) {
191	193		playbook := strings.ReplaceAll(task.Task, " (enable)", "")
192	194		if err := phases.RunPlaybookAndGetResult(admCluster.Kobe, playbook, "", writer); err != nil {
193		-	_ = c.taskLogService.EndDetail(&task, "component", constant.TaskLogStatusFailed, err.Error())
	195	+	_ = c.taskLogService.EndDetail(&task, component.Name, "component", constant.TaskLogStatusFailed, err.Error())
194	196		c.errHandlerComponent(component, constant.StatusFailed, err)
195	197		return
196	198		}
197		-	_ = c.taskLogService.EndDetail(&task, "component", constant.TaskLogStatusSuccess, "")
	199	+	_ = c.taskLogService.EndDetail(&task, component.Name, "component", constant.TaskLogStatusSuccess, "")
198	200		component.Status = constant.StatusWaiting
199	201		if err := db.DB.Save(&component).Error; err != nil {
200	202		logger.Log.Errorf("save component status err: %s", err.Error())
		skipped 25 lines
226	228		return fmt.Errorf("save tasklog failed, err: %v", err)
227	229		}
228	230
229		-	admCluster, writer, err := c.loadAdmCluster(cluster, component, map[string]interface{}{}, constant.StatusDisabled)
230		-	if err != nil {
231		-	return err
232		-	}
233	231		if err := db.DB.Model(&model.ClusterSpecComponent{}).Where("id = ?", component.ID).
234	232		Updates(map[string]interface{}{"status": constant.StatusTerminating, "message": ""}).Error; err != nil {
235	233		return err
236	234		}
237	235
238		-	go c.dodelete(admCluster, writer, task, component)
	236	+	go c.dodelete(&cluster, task, component)
239	237
240	238		return nil
241	239		}
242	240
243		-	func (c componentService) dodelete(admCluster *adm.AnsibleHelper, writer io.Writer, task model.TaskLogDetail, component model.ClusterSpecComponent) {
	241	+	func (c componentService) dodelete(cluster *model.Cluster, task model.TaskLogDetail, component model.ClusterSpecComponent) {
	242	+	admCluster, writer, err := c.loadAdmCluster(*cluster, component, map[string]interface{}{}, constant.StatusDisabled)
	243	+	if err != nil {
	244	+	_ = c.taskLogService.EndDetail(&task, component.Name, "component", constant.TaskLogStatusFailed, err.Error())
	245	+	c.errHandlerComponent(component, constant.StatusFailed, err)
	246	+	return
	247	+	}
244	248		playbook := strings.ReplaceAll(task.Task, " (disable)", "")
245	249		if err := phases.RunPlaybookAndGetResult(admCluster.Kobe, playbook, "", writer); err != nil {
246		-	_ = c.taskLogService.EndDetail(&task, "component", constant.TaskLogStatusFailed, err.Error())
	250	+	_ = c.taskLogService.EndDetail(&task, component.Name, "component", constant.TaskLogStatusFailed, err.Error())
247	251		c.errHandlerComponent(component, constant.StatusFailed, err)
248	252		return
249	253		}
250		-	_ = c.taskLogService.EndDetail(&task, "component", constant.TaskLogStatusSuccess, "")
	254	+	_ = c.taskLogService.EndDetail(&task, component.Name, "component", constant.TaskLogStatusSuccess, "")
251	255		_ = db.DB.Where("id = ?", component.ID).Delete(&model.ClusterSpecComponent{})
252	256		}
253	257
		skipped 160 lines
414	418
415	419		writer, err := ansible.CreateAnsibleLogWriterWithId(cluster.Name, fmt.Sprintf("%s (%s)", component.ID, operation))
416	420		if err != nil {
417		-	return admCluster, writer, err
418		-	}
419		-	if err := db.DB.Model(&model.ClusterSpecComponent{}).Where("id = ?", component.ID).Updates(map[string]interface{}{
420		-	"status": constant.StatusTerminating,
421		-	"message": "",
422		-	}).Error; err != nil {
423	421		return admCluster, writer, err
424	422		}
425	423
		skipped 39 lines

■ ■ ■ ■ ■ ■

pkg/service/kubepi.go

		skipped 7 lines
8	8		"github.com/KubeOperator/KubeOperator/pkg/dto"
9	9		"github.com/KubeOperator/KubeOperator/pkg/logger"
10	10		"github.com/KubeOperator/KubeOperator/pkg/model"
	11	+	"github.com/KubeOperator/KubeOperator/pkg/repository"
11	12		"github.com/KubeOperator/KubeOperator/pkg/util/encrypt"
12	13		"github.com/KubeOperator/KubeOperator/pkg/util/kubepi"
13	14		"github.com/jinzhu/gorm"
		skipped 8 lines
22	23		}
23	24
24	25		func NewKubepiService() KubepiService {
25		-	return &kubepiService{}
	26	+	return &kubepiService{
	27	+	clusterRepo: repository.NewClusterRepository(),
	28	+	}
26	29		}
27	30
28	31		type kubepiService struct {
	32	+	clusterRepo repository.ClusterRepository
29	33		}
30	34
31	35		type ConnInfo struct {
32		-	Name string `josn:"name"`
33		-	Password string `josn:"password"`
	36	+	Name string `josn:"name"`
	37	+	Password string `josn:"password"`
	38	+	Cluster model.Cluster `json:"cluster"`
34	39		}
35	40
36	41		func (c kubepiService) GetKubePiUser() (*kubepi.ListUser, error) {
		skipped 67 lines
104	109		return kubepiClient.CheckLogin()
105	110		}
106	111
107		-	func (s kubepiService) LoadInfo(project, cluster string, isAdmin bool) (ConnInfo, error) {
	112	+	func (s kubepiService) LoadInfo(project, clusterName string, isAdmin bool) (ConnInfo, error) {
	113	+	cluster, err := s.clusterRepo.GetWithPreload(clusterName, []string{"SpecConf", "Secret"})
	114	+	if err != nil {
	115	+	return nil, err
	116	+	}
108	117		var bind model.KubepiBind
109	118		if isAdmin {
110	119		if err := db.DB.Where("source_type = ?", "ADMIN").First(&bind).Error; err != nil {
		skipped 2 lines
113	122		}
114	123		return nil, err
115	124		}
116		-	return &ConnInfo{Name: bind.BindUser, Password: bind.BindPassword}, nil
	125	+	return &ConnInfo{Name: bind.BindUser, Password: bind.BindPassword, Cluster: cluster}, nil
117	126		}
118		-	if err := db.DB.Where("cluster = ? AND source_type = ?", cluster, constant.ResourceCluster).First(&bind).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
	127	+	if err := db.DB.Where("cluster = ? AND source_type = ?", clusterName, constant.ResourceCluster).First(&bind).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
119	128		return nil, err
120	129		}
121	130		if bind.ID != "" {
122		-	return &ConnInfo{Name: bind.BindUser, Password: bind.BindPassword}, nil
	131	+	return &ConnInfo{Name: bind.BindUser, Password: bind.BindPassword, Cluster: cluster}, nil
123	132		}
124	133
125	134		if err := db.DB.Where("project = ? AND source_type = ?", project, constant.ResourceProject).First(&bind).Error; err != nil {
		skipped 2 lines
128	137		}
129	138		return nil, err
130	139		}
131		-	return &ConnInfo{Name: bind.BindUser, Password: bind.BindPassword}, nil
	140	+	return &ConnInfo{Name: bind.BindUser, Password: bind.BindPassword, Cluster: cluster}, nil
132	141		}
133	142

■ ■ ■ ■ ■ ■

pkg/service/msg.go

		skipped 2 lines
3	3		import (
4	4		"bytes"
5	5		"encoding/json"
	6	+	"fmt"
	7	+	"html/template"
	8	+	"io"
	9	+	"reflect"
	10	+
6	11		"github.com/KubeOperator/KubeOperator/bindata"
7	12		"github.com/KubeOperator/KubeOperator/pkg/constant"
8	13		"github.com/KubeOperator/KubeOperator/pkg/db"
		skipped 2 lines
11	16		"github.com/KubeOperator/KubeOperator/pkg/model"
12	17		msgClient "github.com/KubeOperator/KubeOperator/pkg/util/msg"
13	18		"github.com/jinzhu/gorm"
14		-	"html/template"
15		-	"io"
16		-	"reflect"
17	19		)
18	20
19	21		type MsgService interface {
		skipped 50 lines
70	72		content["operator"] = title
71	73		if success {
72	74		msg.Level = constant.MsgInfo
73		-	content["title"] = title + "成功"
	75	+	if name, ok := content["detailName"]; ok {
	76	+	content["title"] = fmt.Sprintf("%s %s 成功", title, name)
	77	+	} else {
	78	+	content["title"] = fmt.Sprintf("%s成功", title)
	79	+	}
74	80		} else {
75	81		msg.Level = constant.MsgWarning
76		-	content["title"] = title + "失败"
	82	+	if name, ok := content["detailName"]; ok {
	83	+	content["title"] = fmt.Sprintf("%s %s 失败", title, name)
	84	+	} else {
	85	+	content["title"] = fmt.Sprintf("%s失败", title)
	86	+	}
77	87		}
78	88		if name == constant.LicenseExpires {
79	89		content["title"] = content["message"]
		skipped 10 lines
90	100		)
91	101
92	102		operate := name
93		-	if name != constant.ClusterInstall && name != constant.LicenseExpires {
	103	+	if scope != constant.System {
94	104		operate = constant.ClusterOperator
95	105		}
96	106		if err := db.DB.Model(model.MsgSubscribe{}).Where("name = ? AND type = ? AND resource_id = ?", operate, msg.Type, resourceId).First(&subscribe).Error; err != nil {
		skipped 164 lines

■ ■ ■ ■ ■ ■

pkg/service/task_log.go

		skipped 31 lines
32	32		IsTaskOn(clusterName string) bool
33	33
34	34		StartDetail(detail *model.TaskLogDetail) error
35		-	EndDetail(detail *model.TaskLogDetail, taskType string, statu string, message string) error
	35	+	EndDetail(detail *model.TaskLogDetail, name, taskType, statu, message string) error
36	36		SaveDetail(detail *model.TaskLogDetail) error
37	37
38	38		RestartTask(cluster *model.Cluster, operation string) error
		skipped 210 lines
249	249		}
250	250
251	251		// taskType component/provisioner
252		-	func (c taskLogService) EndDetail(detail model.TaskLogDetail, taskType string, status string, message string) error {
	252	+	func (c taskLogService) EndDetail(detail model.TaskLogDetail, name, taskType, status, message string) error {
253	253		detail.Status = status
254	254		detail.Message = message
255	255		detail.EndTime = time.Now().Unix()
		skipped 2 lines
258	258		return err
259	259		}
260	260		operation := ""
	261	+	content := make(map[string]string)
	262	+	content["detailName"] = name
261	263		if strings.Contains(detail.Task, "enable") {
262	264		if taskType == "provisioner" {
263	265		operation = constant.ClusterEnableProvisioner
		skipped 8 lines
272	274		}
273	275		}
274	276		if status == constant.TaskLogStatusSuccess {
275		-	_ = c.msgService.SendMsg(operation, constant.Cluster, cluster, true, map[string]string{})
	277	+	_ = c.msgService.SendMsg(operation, constant.Cluster, cluster, true, content)
276	278		} else {
277		-	_ = c.msgService.SendMsg(operation, constant.Cluster, cluster, false, map[string]string{"errMsg": message})
	279	+	content["errMsg"] = message
	280	+	_ = c.msgService.SendMsg(operation, constant.Cluster, cluster, false, content)
278	281		}
279	282
280	283		return db.DB.Save(detail).Error
		skipped 94 lines

Merge branch 'master' of https://github.com/KubeOperator/KubeOperator