feat: basic addon surpport stop/start (#3766)

apistructs/addon.go

@@ -553,6 +553,8 @@ type AddonFetchResponseData struct {
 	CustomAddonType string `json:"customAddonType"`
 	// TenantOwner addon 租户owner的 instancerouting id
 	TenantOwner string `json:"tenantOwner"`
+	// IsInsideAddon addon 是否是 inside addon（如 kafka addon 中的 inside addon 是 zookeeper addon）
+	IsInsideAddon string `json:"isInsideAddon`
 }
 
 // ReferenceInfo 引用信息

apistructs/runtime_dto.go

@@ -213,3 +213,32 @@ type BatchRuntimeReDeployResults struct {
 	UnReDeployedIds []uint64           `json:"reDeployedFailedIds,omitempty"`
 	ErrMsg          []string           `json:"errorMsgs,omitempty"`
 }
+
+// AddonScaleRecords 表示 Addon 的 scale 请求群信息
+type AddonScaleRecords struct {
+	// Addons 不为空则无需设置 AddonRoutingIDs, 二者必选其一
+	// 格式: map[{addon instance's routing ID}]AddonScaleRecord
+	Addons map[string]AddonScaleRecord `json:"addonScaleRecords,omitempty"`
+	// AddonRoutingIDs is the list of addon instance's routing ID
+	AddonRoutingIDs []string `json:"ids,omitempty"`
+}
+
+// AddonScaleRecord is the addon
+type AddonScaleRecord struct {
+	AddonName                   string                                      `json:"addonName,omitempty"`
+	ServiceResourcesAndReplicas map[string]AddonServiceResourcesAndReplicas `json:"services,omitempty"`
+}
+
+// AddonServiceResourcesAndReplicas set the desired resources and replicas for addon services
+type AddonServiceResourcesAndReplicas struct {
+	Resources Resources `json:"resources,omitempty"`
+	Replicas  int32     `json:"replicas,omitempty"`
+}
+
+type AddonScaleResults struct {
+	Total     int `json:"total"`
+	Successed int `json:"successed"`
+	Faild     int `json:"failed"`
+	// FailedInfo   map[{addon routingID}]errMsg
+	FailedInfo map[string]string `json:"errors,omitempty"`
+}

bundle/orchestrator.go

@@ -183,7 +183,7 @@ func (b *Bundle) batchProcessRuntimes(req apistructs.RuntimeScaleRecords, orgID
 	}
 
 	if rsp.Data == nil {
-		return []byte{}, errors.Errorf("return of scale action %s is nil", action)
+		return nil, errors.Errorf("return of scale action %s is nil", action)
 	}
 
 	dataBytes, err := json.Marshal(rsp.Data)
@@ -193,7 +193,6 @@ func (b *Bundle) batchProcessRuntimes(req apistructs.RuntimeScaleRecords, orgID
 
 	return dataBytes, nil
 }
-
 func (b *Bundle) RuntimesClusterReferred(userID, orgID, clusterName string) (referred bool, err error) {
 	host, err := b.urls.Orchestrator()
 	if err != nil {
@@ -223,3 +222,36 @@ func (b *Bundle) RuntimesClusterReferred(userID, orgID, clusterName string) (ref
 
 	return rsp.Data, nil
 }
+
+// ScaleAddon scale down addon (replicas N-->0) or scale up addon (replicas 0-->N)
+func (b *Bundle) ScaleAddon(req apistructs.AddonScaleRecords, orgID uint64, userID string, action string) (*apistructs.AddonScaleResults, error) {
+	host, err := b.urls.Orchestrator()
+	if err != nil {
+		return nil, err
+	}
+	hc := b.hc
+
+	var rsp struct {
+		apistructs.Header
+		Data interface{}
+	}
+
+	resp, err := hc.Post(host).Path(fmt.Sprintf("/api/addons?scale_action=%s", action)).
+		Header(httputil.OrgHeader, strconv.FormatUint(orgID, 10)).
+		Header(httputil.UserHeader, userID).
+		JSONBody(req).Do().JSON(&rsp)
+
+	if err != nil {
+		return nil, apierrors.ErrInvoke.InternalError(err)
+	}
+	if !resp.IsOK() || !rsp.Success {
+		return nil, toAPIError(resp.StatusCode(), rsp.Error)
+	}
+
+	result, ok := rsp.Data.(apistructs.AddonScaleResults)
+	if !ok {
+		return nil, errors.Errorf("the response rsp.Data is not in type apistructs.AddonScaleResults.")
+	}
+
+	return &result, nil
+}

modules/orchestrator/dbclient/addon_instance_routing.go

@@ -436,7 +436,7 @@ func (db *DBClient) GetRoutingInstancesByProject(orgID, projectID uint64, catego
 		Where("project_id = ?", projectID).
 		Where("category != ?", "discovery").
 		Where("is_deleted = ?", apistructs.AddonNotDeleted).
-		Where("status in (?)", []apistructs.AddonStatus{apistructs.AddonAttached, apistructs.AddonAttaching, apistructs.AddonAttachFail})
+		Where("status in (?)", []apistructs.AddonStatus{apistructs.AddonAttached, apistructs.AddonAttaching, apistructs.AddonAttachFail, apistructs.AddonOffline})
 	if category != "" {
 		client = client.Where("category = ?", category)
 	}

modules/orchestrator/endpoints/addon.go

@@ -17,6 +17,7 @@ package endpoints
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"strconv"
 	"strings"
@@ -26,14 +27,14 @@ import (
 	"github.com/sirupsen/logrus"
 
 	"github.com/erda-project/erda/apistructs"
+	"github.com/erda-project/erda/modules/orchestrator/services/addon"
+	"github.com/erda-project/erda/modules/orchestrator/services/apierrors"
+	"github.com/erda-project/erda/modules/orchestrator/utils"
 	"github.com/erda-project/erda/modules/pkg/user"
 	"github.com/erda-project/erda/pkg/http/httpserver"
 	"github.com/erda-project/erda/pkg/http/httputil"
 	"github.com/erda-project/erda/pkg/parser/diceyml"
 	"github.com/erda-project/erda/pkg/strutil"
-
-	"github.com/erda-project/erda/modules/orchestrator/services/addon"
-	"github.com/erda-project/erda/modules/orchestrator/services/apierrors"
 )
 
 func (e *Endpoints) CreateAddonDirectly(ctx context.Context, r *http.Request, vars map[string]string) (httpserver.Responser, error) {
@@ -194,6 +195,69 @@ func (e *Endpoints) GetAddon(ctx context.Context, r *http.Request, vars map[stri
 	return httpserver.OkResp(addonInfo)
 }
 
+// ScaleAddon 设置 addon 服务停止（副本数 N--->0）/启动（副本数 0--->N）
+func (e *Endpoints) ScaleAddon(ctx context.Context, r *http.Request, vars map[string]string) (httpserver.Responser, error) {
+
+	action := r.URL.Query().Get(apistructs.ScaleAction)
+	if action != apistructs.ScaleActionUp && action != apistructs.ScaleActionDown {
+		return apierrors.ErrScaleAddon.InvalidParameter("no parameter " + apistructs.ScaleAction + " or invalid parameter value for parameter " + apistructs.ScaleAction).ToResp(), nil
+	}
+
+	userID, err := user.GetUserID(r)
+	if err != nil {
+		return apierrors.ErrScaleAddon.NotLogin().ToResp(), nil
+	}
+
+	orgID := r.Header.Get(httputil.OrgHeader)
+	if orgID == "" {
+		return apierrors.ErrScaleAddon.MissingParameter("ORG-ID").ToResp(), nil
+	}
+
+	var addonScaleRecords apistructs.AddonScaleRecords
+	if err := json.NewDecoder(r.Body).Decode(&addonScaleRecords); err != nil {
+		return utils.ErrRespIllegalParam(err, "failed to batch update Overlay, failed to parse req")
+	}
+
+	if len(addonScaleRecords.Addons) == 0 && len(addonScaleRecords.AddonRoutingIDs) == 0 {
+		return utils.ErrRespIllegalParam(err, "failed to batch update Overlay, no addonScaleRecords or ids provided in request body")
+	}
+
+	if len(addonScaleRecords.Addons) != 0 && len(addonScaleRecords.AddonRoutingIDs) != 0 {
+		return utils.ErrRespIllegalParam(err, "failed to batch update Overlay, addonScaleRecords and ids must only one with non-empty values in request body")
+	}
+
+	var result apistructs.AddonScaleResults
+	if len(addonScaleRecords.AddonRoutingIDs) > 0 {
+		// scale addon from N--->0 or 0--->N
+		result.Total = len(addonScaleRecords.AddonRoutingIDs)
+		for _, addonRoutingId := range addonScaleRecords.AddonRoutingIDs {
+			if err := e.addon.Scale(userID.String(), addonRoutingId, action); err != nil {
+				logrus.Errorf("Do %s for addon %s failed. error: %v", action, addonRoutingId, err)
+				msg := fmt.Sprintf("Do %s for addon %s failed. error: %v", action, addonRoutingId, err)
+				if result.FailedInfo == nil {
+					result.FailedInfo = make(map[string]string)
+				}
+				result.Faild++
+				result.FailedInfo[addonRoutingId] = msg
+			} else {
+				logrus.Infof("Do %s for addon %s successfully.", action, vars["addonID"])
+				result.Successed++
+			}
+		}
+	}
+
+	if len(addonScaleRecords.Addons) > 0 {
+		// ToDo: scale addon for replicas or/and resources
+	}
+
+	if result.Faild > 0 {
+		return httpserver.NotOkResp(result, http.StatusInternalServerError)
+	}
+	logrus.Infof("scale all addons successfully")
+
+	return httpserver.OkResp(result)
+}
+
 // DeleteAddon 删除 addon
 func (e *Endpoints) DeleteAddon(ctx context.Context, r *http.Request, vars map[string]string) (httpserver.Responser, error) {
 	userID, err := user.GetUserID(r)

modules/orchestrator/endpoints/endpoints.go

@@ -262,6 +262,7 @@ func (e *Endpoints) Routes() []httpserver.Endpoint {
 		{Path: "/api/addons/{addonID}", Method: http.MethodGet, Handler: e.GetAddon},
 		{Path: "/api/addons/{addonID}/actions/references", Method: http.MethodGet, Handler: e.GetAddonReferences},
 		{Path: "/api/addons", Method: http.MethodGet, Handler: e.ListAddon},
+		{Path: "/api/addons", Method: http.MethodPost, Handler: e.ScaleAddon},
 		{Path: "/api/addons/actions/list-extension", Method: http.MethodGet, Handler: e.ListExtensionAddon},
 		{Path: "/api/addons/types/{addonName}", Method: http.MethodGet, Handler: e.ListByAddonName},
 		{Path: "/api/addons/actions/list-available", Method: http.MethodGet, Handler: e.ListAvailableAddon},