feat: [controller & cli] support cancel agent upgrade

cli/ctl/agent.go

@@ -136,9 +136,16 @@ func RegisterAgentUpgradeCommand() *cobra.Command {
 		Use:   "agent-upgrade",
 		Short: "agent upgrade operation commands",
 		Example: "deepflow-ctl agent-upgrade list\n" +
-			"deepflow-ctl agent-upgrade agent-name --image-name=deepflow-agent\n",
+			"deepflow-ctl agent-upgrade agent-name --image-name=deepflow-agent\n" +
+			"deepflow-ctl agent-upgrade cancel agent-name\n",
 		Run: func(cmd *cobra.Command, args []string) {
-			if len(args) == 1 {
+			if len(args) == 2 {
+				if args[0] == "cancel" {
+					cancelUpgadeAgent(cmd, args)
+				} else {
+					fmt.Println(cmd.Example)
+				}
+			} else if len(args) == 1 {
 				if args[0] == "list" {
 					listAgentUpgrade(cmd, args)
 				} else if imageName != "" {
@@ -482,6 +489,80 @@ func upgadeAgent(cmd *cobra.Command, args []string) {
 	}
 }
 
+func cancelUpgadeAgent(cmd *cobra.Command, args []string) {
+	if len(args) != 2 {
+		fmt.Fprintf(os.Stderr, "must specify name. Examples: \n%s", cmd.Example)
+		return
+	}
+	vtapName := args[1]
+
+	server := common.GetServerInfo(cmd)
+	serverURL := fmt.Sprintf("http://%s:%d/v1/controllers/", server.IP, server.Port)
+	response, err := common.CURLPerform("GET", serverURL, nil, "",
+		[]common.HTTPOption{common.WithTimeout(common.GetTimeout(cmd)), common.WithORGID(common.GetORGID(cmd))}...)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		return
+	}
+	controllerArray := response.Get("DATA").MustArray()
+	hosts := map[string]struct{}{
+		server.IP: struct{}{},
+	}
+	if len(controllerArray) > 0 {
+		for index, _ := range controllerArray {
+			nodeType := response.Get("DATA").GetIndex(index).Get("NODE_TYPE").MustInt()
+			ip := response.Get("DATA").GetIndex(index).Get("IP").MustString()
+			if nodeType == 1 && ip != "" {
+				hosts[ip] = struct{}{}
+			}
+		}
+	} else {
+		fmt.Printf("get server info failed, url: %s\n", serverURL)
+		return
+	}
+
+	vtapURL := fmt.Sprintf("http://%s:%d/v1/vtaps/?name=%s", server.IP, server.Port, vtapName)
+	response, err = common.CURLPerform("GET", vtapURL, nil, "",
+		[]common.HTTPOption{common.WithTimeout(common.GetTimeout(cmd)), common.WithORGID(common.GetORGID(cmd))}...)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		return
+	}
+
+	var (
+		vtapController string
+		vtapLcuuid     string
+	)
+
+	if len(response.Get("DATA").MustArray()) > 0 {
+		vtapLcuuid = response.Get("DATA").GetIndex(0).Get("LCUUID").MustString()
+		vtapController = response.Get("DATA").GetIndex(0).Get("CONTROLLER_IP").MustString()
+	} else {
+		fmt.Printf("get agent(%s) info failed, url: %s\n", vtapName, vtapURL)
+		return
+	}
+	if vtapController == "" || vtapLcuuid == "" {
+		fmt.Printf("get agent(%s) info failed, url: %s\n", vtapName, vtapURL)
+		return
+	}
+
+	hosts[vtapController] = struct{}{}
+	url_format := "http://%s:%d/v1/cancel-upgrade/vtap/%s/"
+	body := map[string]interface{}{}
+	for host, _ := range hosts {
+		url := fmt.Sprintf(url_format, host, server.Port, vtapLcuuid)
+		response, err := common.CURLPerform("PATCH", url, body, "",
+			[]common.HTTPOption{common.WithTimeout(common.GetTimeout(cmd)), common.WithORGID(common.GetORGID(cmd))}...)
+		if err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			fmt.Printf("cancel upgrade agent %s server %s failed, response: %s\n", vtapName, host, response)
+			continue
+		} else {
+			fmt.Printf("cancel agent %s upgrade to server(%s) success\n", vtapName, host)
+		}
+	}
+}
+
 func rebalance(cmd *cobra.Command, rebalanceType RebalanceType, typeVal string) error {
 	if isDebug {
 		return rebalanceDebug(cmd, typeVal)

server/controller/trisolaris/services/grpc/agentsynchronize/upgrade.go

@@ -138,6 +138,12 @@ func (e *UpgradeEvent) Upgrade(r *api.UpgradeRequest, in api.Synchronizer_Upgrad
 				log.Errorf("vtap(%s) teamID:%s-%d, err:%s", vtapCacheKey, teamIDStr, teamIDInt, err, logger.NewORGPrefix(orgID))
 				break
 			}
+
+			// if upgrade is canceled/completed, should close stream
+			if vtapCache.GetExpectedRevision() == "" {
+				log.Warningf("vtap(%s) teamID:%s-%d upgrade is canceled/completed", vtapCacheKey, teamIDStr, teamIDInt, logger.NewORGPrefix(orgID))
+				break
+			}
 		}
 	}
 

server/controller/trisolaris/services/grpc/synchronize/upgrade.go

@@ -138,6 +138,12 @@ func (e *UpgradeEvent) Upgrade(r *api.UpgradeRequest, in api.Synchronizer_Upgrad
 				log.Errorf("vtap(%s) teamID:%s-%d, err:%s", vtapCacheKey, teamIDStr, teamIDInt, err, logger.NewORGPrefix(orgID))
 				break
 			}
+
+			// if upgrade is canceled/completed, should close stream
+			if vtapCache.GetExpectedRevision() == "" {
+				log.Warningf("vtap(%s) teamID:%s-%d upgrade is canceled/completed", vtapCacheKey, teamIDStr, teamIDInt, logger.NewORGPrefix(orgID))
+				break
+			}
 		}
 	}
 

server/controller/trisolaris/services/http/upgrade/upgrade.go

@@ -101,10 +101,51 @@ func Upgrade(c *gin.Context) {
 		return
 	}
 	vTapCache.UpdateUpgradeInfo(expectedRevision, upgradeInfo.ImageName)
-	log.Infof("vtap(%s, %s) upgrade:(%s, %s)", orgIDInt, vtap.Name, key, expectedRevision, upgradeInfo.ImageName, logger.NewORGPrefix(orgIDInt))
+	log.Infof("vtap(%s, %s) upgrade:(%s, %s)", vtap.Name, key, expectedRevision, upgradeInfo.ImageName)
+	common.Response(c, nil, common.NewReponse("SUCCESS", "", nil, ""))
+}
+
+func CancelUpgrade(c *gin.Context) {
+	var err error
+	lcuuid := c.Param("lcuuid")
+	if lcuuid == "" {
+		common.Response(c, nil, common.NewReponse("FAILED", "", nil, "not find lcuuid param"))
+		return
+	}
+	orgID, _ := c.Get(HEADER_KEY_X_ORG_ID)
+	orgIDInt := orgID.(int)
+	db, err := mysql.GetDB(orgIDInt)
+	if err != nil {
+		common.Response(c, nil, common.NewReponse("FAILED", "", nil, err.Error()))
+		return
+	}
+
+	vtap, err := dbmgr.DBMgr[models.VTap](db.DB).GetFromLcuuid(lcuuid)
+	if err != nil {
+		log.Error(err)
+		common.Response(c, nil, common.NewReponse("FAILED", "", nil, fmt.Sprintf("orgID=%d, %s", orgIDInt, err)))
+		return
+	}
+	key := vtap.CtrlIP + "-" + vtap.CtrlMac
+	vTapCache := trisolaris.GetORGVTapInfo(orgIDInt).GetVTapCache(key)
+	if vTapCache == nil {
+		common.Response(c, nil, common.NewReponse("FAILED", "", nil, fmt.Sprintf("orgID=%d, not found vtap cache", orgIDInt)))
+		return
+	}
+
+	// if upgrade is completed, should return error message
+	if vTapCache.GetExpectedRevision() == "" || vTapCache.GetExpectedRevision() == vTapCache.GetRevision() {
+		common.Response(c, nil, common.NewReponse("FAILED", "", nil, fmt.Sprintf("orgID=%d, vtap(%s, %s) upgrade is completed, unable to cancel", orgIDInt, key)))
+		return
+	}
+
+	// cancel upgrade
+	vTapCache.UpdateUpgradeInfo("", "")
+	log.Infof("vtap(%s, %s) upgrade is canceled", vtap.Name, key)
 	common.Response(c, nil, common.NewReponse("SUCCESS", "", nil, ""))
 }
 
 func (*UpgradeService) Register(mux *gin.Engine) {
 	mux.PATCH("v1/upgrade/vtap/:lcuuid/", Upgrade)
+	mux.PATCH("v1/cancel-upgrade/vtap/:lcuuid/", CancelUpgrade)
 }