refresh xpa task

api/handler/service.go

@@ -1916,6 +1916,20 @@ func (s *ServiceAction) AddAutoscalerRule(req *api_model.AutoscalerRuleReq) erro
 		}
 	}
 
+	taskbody := map[string]interface{}{
+		"service_id": r.ServiceID,
+		"rule_id":    r.RuleID,
+	}
+	if err := s.MQClient.SendBuilderTopic(gclient.TaskStruct{
+		TaskType: "refreshhpa",
+		TaskBody: taskbody,
+		Topic:    gclient.WorkerTopic,
+	}); err != nil {
+		logrus.Errorf("send 'refreshhpa' task: %v", err)
+		return err
+	}
+	logrus.Infof("rule id: %s; successfully send 'refreshhpa' task.", r.RuleID)
+
 	return tx.Commit().Error
 }
 
@@ -1953,6 +1967,20 @@ func (s *ServiceAction) UpdAutoscalerRule(req *api_model.AutoscalerRuleReq) erro
 		}
 	}
 
+	taskbody := map[string]interface{}{
+		"service_id": rule.ServiceID,
+		"rule_id":    rule.RuleID,
+	}
+	if err := s.MQClient.SendBuilderTopic(gclient.TaskStruct{
+		TaskType: "refreshhpa",
+		TaskBody: taskbody,
+		Topic:    gclient.WorkerTopic,
+	}); err != nil {
+		logrus.Errorf("send 'refreshhpa' task: %v", err)
+		return err
+	}
+	logrus.Infof("rule id: %s; successfully send 'refreshhpa' task.", rule.RuleID)
+
 	return tx.Commit().Error
 }
 

db/dao/dao.go

@@ -457,6 +457,7 @@ type TenantServceAutoscalerRulesDao interface {
 	Dao
 	GetByRuleID(ruleID string) (*model.TenantServiceAutoscalerRules, error)
 	ListByServiceID(serviceID string) ([]*model.TenantServiceAutoscalerRules, error)
+	ListEnableOnesByServiceID(serviceID string) ([]*model.TenantServiceAutoscalerRules, error)
 }
 
 // TenantServceAutoscalerRuleMetricsDao -

db/mysql/dao/tenants.go

@@ -1506,6 +1506,15 @@ func (t *TenantServceAutoscalerRulesDaoImpl) ListByServiceID(serviceID string) (
 	return rules, nil
 }
 
+// ListEnableOnesByServiceID -
+func (t *TenantServceAutoscalerRulesDaoImpl) ListEnableOnesByServiceID(serviceID string) ([]*model.TenantServiceAutoscalerRules, error) {
+	var rules []*model.TenantServiceAutoscalerRules
+	if err := t.DB.Where("service_id=? and enable=?", serviceID, true).Find(&rules).Error; err != nil {
+		return nil, err
+	}
+	return rules, nil
+}
+
 // TenantServceAutoscalerRuleMetricsDaoImpl -
 type TenantServceAutoscalerRuleMetricsDaoImpl struct {
 	DB *gorm.DB

worker/appm/controller/controller.go

@@ -61,6 +61,9 @@ var TypeApplyRuleController TypeController = "apply_rule"
 // TypeApplyConfigController -
 var TypeApplyConfigController TypeController = "apply_config"
 
+// TypeControllerRefreshHPA -
+var TypeControllerRefreshHPA TypeController = "refreshhpa"
+
 //Manager controller manager
 type Manager struct {
 	ctx         context.Context
@@ -150,6 +153,13 @@ func (m *Manager) StartController(controllerType TypeController, apps ...v1.AppS
 			manager:      m,
 			stopChan:     make(chan struct{}),
 		}
+	case TypeControllerRefreshHPA:
+		controller = &refreshXPAController{
+			controllerID: controllerID,
+			appService:   apps,
+			manager:      m,
+			stopChan:     make(chan struct{}),
+		}
 	default:
 		return fmt.Errorf("No support controller")
 	}

worker/appm/controller/refresh_xpa.go

@@ -0,0 +1,75 @@
+// RAINBOND, Application Management Platform
+// Copyright (C) 2014-2017 Goodrain Co., Ltd.
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version. For any non-GPL usage of Rainbond,
+// one or multiple Commercial Licenses authorized by Goodrain Co., Ltd.
+// must be obtained first.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package controller
+
+import (
+	"sync"
+
+	"github.com/Sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/goodrain/rainbond/worker/appm/f"
+	"github.com/goodrain/rainbond/worker/appm/types/v1"
+)
+
+type refreshXPAController struct {
+	controllerID string
+	appService   []v1.AppService
+	manager      *Manager
+	stopChan     chan struct{}
+}
+
+// Begin begins applying rule
+func (a *refreshXPAController) Begin() {
+	var wait sync.WaitGroup
+	for _, service := range a.appService {
+		go func(service v1.AppService) {
+			wait.Add(1)
+			defer wait.Done()
+			if err := a.applyOne(a.manager.client, &service); err != nil {
+				logrus.Errorf("apply rules for service %s failure: %s", service.ServiceAlias, err.Error())
+			}
+		}(service)
+	}
+	wait.Wait()
+	a.manager.callback(a.controllerID, nil)
+}
+
+func (a *refreshXPAController) applyOne(clientset *kubernetes.Clientset, app *v1.AppService) error {
+	for _, hpa := range app.GetHPAs() {
+		f.EnsureHPA(hpa, clientset)
+	}
+
+	for _, hpa := range app.GetDelHPAs() {
+		logrus.Debugf("hpa name: %s; start deleting hpa.", hpa.GetName())
+		err := clientset.AutoscalingV2beta1().HorizontalPodAutoscalers(hpa.GetNamespace()).Delete(hpa.GetName(), &metav1.DeleteOptions{})
+		if err != nil {
+			// don't return error, hope it is ok next time
+			logrus.Warningf("error deleting secret(%#v): %v", hpa, err)
+		}
+	}
+
+	return nil
+}
+
+func (a *refreshXPAController) Stop() error {
+	close(a.stopChan)
+	return nil
+}

worker/appm/conversion/autoscaler.go

@@ -45,6 +45,7 @@ func TenantServiceAutoscaler(as *v1.AppService, dbmanager db.Manager) error {
 	if err != nil {
 		return fmt.Errorf("create HPAs: %v", err)
 	}
+	logrus.Debugf("the numbers of HPAs: %d", len(hpas))
 
 	as.SetHPAs(hpas)
 
@@ -52,7 +53,7 @@ func TenantServiceAutoscaler(as *v1.AppService, dbmanager db.Manager) error {
 }
 
 func newHPAs(as *v1.AppService, dbmanager db.Manager) ([]*v2beta1.HorizontalPodAutoscaler, error) {
-	xpaRules, err := dbmanager.TenantServceAutoscalerRulesDao().ListByServiceID(as.ServiceID)
+	xpaRules, err := dbmanager.TenantServceAutoscalerRulesDao().ListEnableOnesByServiceID(as.ServiceID)
 	if err != nil {
 		return nil, err
 	}

worker/appm/f/function.go

@@ -20,13 +20,16 @@ package f
 
 import (
 	"fmt"
+
 	"github.com/Sirupsen/logrus"
-	"github.com/goodrain/rainbond/worker/appm/types/v1"
+	v2beta1 "k8s.io/api/autoscaling/v2beta1"
 	corev1 "k8s.io/api/core/v1"
 	extensions "k8s.io/api/extensions/v1beta1"
 	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
+
+	"github.com/goodrain/rainbond/worker/appm/types/v1"
 )
 
 // ApplyOne applies one rule.
@@ -139,6 +142,7 @@ func ensureSecret(secret *corev1.Secret, clientSet kubernetes.Interface) {
 		logrus.Warningf("error updating secret %+v: %v", secret, err)
 	}
 }
+
 // EnsureEndpoints creates or updates endpoints.
 func EnsureEndpoints(ep *corev1.Endpoints, clientSet kubernetes.Interface) {
 	oldEP, err := clientSet.CoreV1().Endpoints(ep.Namespace).Get(ep.Name, metav1.GetOptions{})
@@ -165,6 +169,27 @@ func EnsureService(new *corev1.Service, clientSet kubernetes.Interface) {
 	ensureService(new, clientSet)
 }
 
+// EnsureHPA -
+func EnsureHPA(new *v2beta1.HorizontalPodAutoscaler, clientSet kubernetes.Interface) {
+	_, err := clientSet.AutoscalingV2beta1().HorizontalPodAutoscalers(new.Namespace).Get(new.Name, metav1.GetOptions{})
+	if err != nil {
+		if k8sErrors.IsNotFound(err) {
+			_, err = clientSet.AutoscalingV2beta1().HorizontalPodAutoscalers(new.Namespace).Create(new)
+			if err != nil {
+				logrus.Warningf("error creating hpa %+v: %v", new, err)
+			}
+			return
+		}
+		logrus.Errorf("error getting hpa(%s): %v", fmt.Sprintf("%s/%s", new.Namespace, new.Name), err)
+		return
+	}
+	_, err = clientSet.AutoscalingV2beta1().HorizontalPodAutoscalers(new.Namespace).Update(new)
+	if err != nil {
+		logrus.Warningf("error updating hpa %+v: %v", new, err)
+		return
+	}
+}
+
 // UpgradeIngress is used to update *extensions.Ingress.
 func UpgradeIngress(clientset *kubernetes.Clientset,
 	as *v1.AppService,
@@ -280,9 +305,9 @@ func UpgradeEndpoints(clientset *kubernetes.Clientset,
 	}
 	for _, n := range new {
 		if o, ok := oldMap[n.Name]; ok {
-			oldEndpoint, err := clientset.CoreV1().Endpoints(n.Namespace).Get(n.Name,metav1.GetOptions{})
+			oldEndpoint, err := clientset.CoreV1().Endpoints(n.Namespace).Get(n.Name, metav1.GetOptions{})
 			if err != nil {
-				if k8sErrors.IsNotFound(err){
+				if k8sErrors.IsNotFound(err) {
 					_, err := clientset.CoreV1().Endpoints(n.Namespace).Create(n)
 					if err != nil {
 						if err := handleErr(fmt.Sprintf("error creating endpoints: %+v: err: %v",
@@ -292,7 +317,7 @@ func UpgradeEndpoints(clientset *kubernetes.Clientset,
 						continue
 					}
 				}
-				if e := handleErr(fmt.Sprintf("err get endpoint[%s:%s], err: %+v", n.Namespace, n.Name, err), err); err != nil{
+				if e := handleErr(fmt.Sprintf("err get endpoint[%s:%s], err: %+v", n.Namespace, n.Name, err), err); err != nil {
 					return e
 				}
 			}
@@ -335,4 +360,3 @@ func UpgradeEndpoints(clientset *kubernetes.Clientset,
 	}
 	return nil
 }
-

worker/appm/types/v1/v1.go

@@ -104,6 +104,7 @@ type AppService struct {
 	statefulset  *v1.StatefulSet
 	deployment   *v1.Deployment
 	hpas         []*v2beta1.HorizontalPodAutoscaler
+	delHPAs      []*v2beta1.HorizontalPodAutoscaler
 	replicasets  []*v1.ReplicaSet
 	services     []*corev1.Service
 	delServices  []*corev1.Service
@@ -541,6 +542,18 @@ func (a *AppService) SetDeletedResources(old *AppService) {
 			a.delServices = append(a.delServices, o)
 		}
 	}
+	for _, o := range old.GetHPAs() {
+		del := true
+		for _, n := range a.GetHPAs() {
+			if o.Name == n.Name {
+				del = false
+				break
+			}
+		}
+		if del {
+			a.delHPAs = append(a.delHPAs, o)
+		}
+	}
 }
 
 // DistinguishPod uses replica set to distinguish between old and new pods
@@ -576,6 +589,11 @@ func (a *AppService) GetHPAs() []*v2beta1.HorizontalPodAutoscaler {
 	return a.hpas
 }
 
+// GetDelHPAs -
+func (a *AppService) GetDelHPAs() []*v2beta1.HorizontalPodAutoscaler {
+	return a.delHPAs
+}
+
 // DelHPA -
 func (a *AppService) DelHPA(hpa *v2beta1.HorizontalPodAutoscaler) {
 	if len(a.hpas) == 0 {

worker/discover/model/model.go

@@ -151,6 +151,13 @@ func NewTaskBody(taskType string, body []byte) TaskBody {
 			return nil
 		}
 		return b
+	case "refreshhpa":
+		b := &RefreshHPATaskBody{}
+		err := ffjson.Unmarshal(body, &b)
+		if err != nil {
+			return nil
+		}
+		return b
 	default:
 		return DefaultTaskBody{}
 	}
@@ -181,6 +188,8 @@ func CreateTaskBody(taskType string) TaskBody {
 		return ApplyPluginConfigTaskBody{}
 	case "delete_tenant":
 		return DeleteTenantTaskBody{}
+	case "refreshhpa":
+		return RefreshHPATaskBody{}
 	default:
 		return DefaultTaskBody{}
 	}
@@ -324,5 +333,12 @@ type DeleteTenantTaskBody struct {
 	TenantID string `json:"tenant_id"`
 }
 
+// RefreshHPATaskBody -
+type RefreshHPATaskBody struct {
+	ServiceID string `json:"service_id"`
+	RuleID    string `json:"rule_id"`
+	EventID   string `json:"eventID"`
+}
+
 //DefaultTaskBody 默认操作任务主体
 type DefaultTaskBody map[string]interface{}

worker/handle/manager.go

@@ -119,6 +119,9 @@ func (m *Manager) AnalystToExec(task *model.Task) error {
 	case "delete_tenant":
 		logrus.Info("start a 'delete_tenant' task worker")
 		return m.deleteTenant(task)
+	case "refreshhpa":
+		logrus.Info("start a 'refreshhpa' task worker")
+		return m.ExecRefreshHPATask(task)
 	default:
 		logrus.Warning("task can not execute because no type is identified")
 		return nil
@@ -457,7 +460,7 @@ func (m *Manager) deleteTenant(task *model.Task) (err error) {
 		var tenant *dbmodel.Tenants
 		tenant, err = db.GetManager().TenantDao().GetTenantByUUID(body.TenantID)
 		if err != nil {
-			err = fmt.Errorf("tenant id: %s; find tenant: %v", err)
+			err = fmt.Errorf("tenant id: %s; find tenant: %v", body.TenantID, err)
 			return
 		}
 		tenant.Status = dbmodel.TenantStatusDeleteFailed.String()
@@ -483,3 +486,42 @@ func (m *Manager) deleteTenant(task *model.Task) (err error) {
 
 	return
 }
+
+// ExecRefreshHPATask executes a 'refresh hpa' task.
+func (m *Manager) ExecRefreshHPATask(task *model.Task) error {
+	body, ok := task.Body.(*model.RefreshHPATaskBody)
+	if !ok {
+		logrus.Errorf("exec task 'refreshhpa'; wrong type: %v", reflect.TypeOf(task))
+		return fmt.Errorf("exec task 'refreshhpa': wrong input")
+	}
+
+	logger := event.GetManager().GetLogger(body.EventID)
+
+	oldAppService := m.store.GetAppService(body.ServiceID)
+	if oldAppService != nil && oldAppService.IsClosed() {
+		logger.Info("application is closed, ignore task 'refreshhpa'", event.GetLastLoggerOption())
+		event.GetManager().ReleaseLogger(logger)
+		return nil
+	}
+
+	newAppService, err := conversion.InitAppService(m.dbmanager, body.ServiceID, nil)
+	if err != nil {
+		logrus.Errorf("Application init create failure:%s", err.Error())
+		logger.Error("Application init create failure", event.GetCallbackLoggerOption())
+		event.GetManager().ReleaseLogger(logger)
+		return fmt.Errorf("Application init create failure")
+	}
+	newAppService.Logger = logger
+	newAppService.SetDeletedResources(oldAppService)
+
+	err = m.controllerManager.StartController(controller.TypeControllerRefreshHPA, *newAppService)
+	if err != nil {
+		logrus.Errorf("Application run  refreshhpa controller failure: %s", err.Error())
+		logger.Error("Application run refreshhpa controller failure", event.GetCallbackLoggerOption())
+		event.GetManager().ReleaseLogger(logger)
+		return fmt.Errorf("refresh hpa: %v", err)
+	}
+
+	logrus.Infof("rule id: %s; successfully refresh hpa", body.RuleID)
+	return nil
+}