api/handler/service_batch_operation.go

Rama main · réplica de lectura del repositorio original.

// RAINBOND, Application Management Platform
// Copyright (C) 2014-2019 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 handler

import (
	"container/list"
	"context"
	"fmt"
	"github.com/goodrain/rainbond/pkg/component/grpc"
	"github.com/goodrain/rainbond/pkg/component/mq"
	"strings"
	"time"

	"github.com/goodrain/rainbond/api/model"
	apiutil "github.com/goodrain/rainbond/api/util"
	"github.com/goodrain/rainbond/db"
	dbmodel "github.com/goodrain/rainbond/db/model"
	gclient "github.com/goodrain/rainbond/mq/client"
	"github.com/goodrain/rainbond/util"
	"github.com/goodrain/rainbond/util/retryutil"
	"github.com/goodrain/rainbond/worker/client"
	"github.com/jinzhu/gorm"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

// BatchOperationHandler batch operation handler
type BatchOperationHandler struct {
	mqCli            gclient.MQClient
	operationHandler *OperationHandler
	statusCli        *client.AppRuntimeSyncClient
	DryRun           bool
	HelmChart        *model.HelmChart
	EventIDs         []string
}

// BatchOperationResult batch operation result
type BatchOperationResult struct {
	BatchResult []OperationResult `json:"batche_result"`
}

// CreateBatchOperationHandler create batch operation handler
func CreateBatchOperationHandler(operationHandler *OperationHandler) *BatchOperationHandler {
	return &BatchOperationHandler{
		mqCli:            mq.Default().MqClient,
		operationHandler: operationHandler,
		statusCli:        grpc.Default().StatusClient,
	}
}

func (b *BatchOperationHandler) serviceStartupSequence(serviceIDs []string) map[string][]string {
	sd, err := NewServiceDependency(serviceIDs)
	if err != nil {
		logrus.Warningf("create a new ServiceDependency: %v", err)
	}
	startupSeqConfigs := sd.serviceStartupSequence()
	logrus.Debugf("startup sequence configurations: %#v", startupSeqConfigs)
	return startupSeqConfigs
}

// Build build
func (b *BatchOperationHandler) Build(ctx context.Context, tenant *dbmodel.Tenants, operator string, batchOpReqs model.BatchOpRequesters) (model.BatchOpResult, error) {
	if logrus.IsLevelEnabled(logrus.DebugLevel) {
		defer util.Elapsed("[BatchOperationHandler] build components")()
	}

	// setup start sequence config
	componentIDs := batchOpReqs.ComponentIDs()
	startupSeqConfigs := b.serviceStartupSequence(componentIDs)

	// check allocatable memory
	allocm, err := NewAllocMemory(ctx, b.statusCli, tenant, batchOpReqs)
	if err != nil {
		return nil, errors.WithMessage(err, "new alloc memory")
	}
	batchOpResult := allocm.BatchOpResult()
	validBuilds := allocm.BatchOpRequests()

	batchOpReqs, batchOpResult2 := b.checkEvents(batchOpReqs)
	batchOpResult = append(batchOpResult, batchOpResult2...)

	// create events
	if err := b.createEvents(tenant.UUID, operator, batchOpReqs, allocm.badOpRequest, allocm.memoryType); err != nil {
		return nil, err
	}
	for i, build := range validBuilds {
		b.operationHandler.SetHelmParameter(b.DryRun, b.HelmChart, b.EventIDs, i == len(validBuilds)-1)
		build.UpdateConfig("boot_seq_dep_service_ids", strings.Join(startupSeqConfigs[build.GetComponentID()], ","))
		err := retryutil.Retry(1*time.Microsecond, 1, func() (bool, error) {
			if err := b.operationHandler.build(build); err != nil {
				return false, err
			}
			return true, nil
		})
		item := build.BatchOpFailureItem()
		if err != nil {
			item.ErrMsg = err.Error()
		} else {
			item.Success()
		}
		batchOpResult = append(batchOpResult, item)
	}

	return batchOpResult, nil
}

// Start batch start
func (b *BatchOperationHandler) Start(ctx context.Context, tenant *dbmodel.Tenants, operator string, batchOpReqs model.BatchOpRequesters) (model.BatchOpResult, error) {
	if logrus.IsLevelEnabled(logrus.DebugLevel) {
		defer util.Elapsed("[BatchOperationHandler] start components")()
	}

	// setup start sequence config
	componentIDs := batchOpReqs.ComponentIDs()
	startupSeqConfigs := b.serviceStartupSequence(componentIDs)

	// chekc allocatable memory
	allocm, err := NewAllocMemory(ctx, b.statusCli, tenant, batchOpReqs)
	if err != nil {
		return nil, errors.WithMessage(err, "new alloc memory")
	}
	batchOpResult := allocm.BatchOpResult()
	validRequestes := allocm.BatchOpRequests()

	batchOpReqs, batchOpResult2 := b.checkEvents(batchOpReqs)
	batchOpResult = append(batchOpResult, batchOpResult2...)

	// create events
	if err := b.createEvents(tenant.UUID, operator, batchOpReqs, allocm.BadOpRequests(), allocm.memoryType); err != nil {
		return nil, err
	}

	for _, req := range validRequestes {
		// startup sequence
		req.UpdateConfig("boot_seq_dep_service_ids", strings.Join(startupSeqConfigs[req.GetComponentID()], ","))
		err := retryutil.Retry(1*time.Microsecond, 1, func() (bool, error) {
			if err := b.operationHandler.Start(req); err != nil {
				return false, err
			}
			return true, nil
		})
		item := req.BatchOpFailureItem()
		if err != nil {
			item.ErrMsg = err.Error()
		} else {
			item.Success()
		}
		batchOpResult = append(batchOpResult, item)
	}

	return batchOpResult, nil
}

// Stop batch stop
func (b *BatchOperationHandler) Stop(ctx context.Context, tenant *dbmodel.Tenants, operator string, batchOpReqs model.BatchOpRequesters) (model.BatchOpResult, error) {
	if logrus.IsLevelEnabled(logrus.DebugLevel) {
		defer util.Elapsed("[BatchOperationHandler] stop components")()
	}

	batchOpReqs, batchOpResult := b.checkEvents(batchOpReqs)

	// create events
	if err := b.createEvents(tenant.UUID, operator, batchOpReqs, nil, ""); err != nil {
		return nil, err
	}

	for _, req := range batchOpReqs {
		err := retryutil.Retry(1*time.Microsecond, 1, func() (bool, error) {
			if err := b.operationHandler.Stop(req); err != nil {
				return false, err
			}
			return true, nil
		})
		item := req.BatchOpFailureItem()
		if err != nil {
			item.ErrMsg = err.Error()
		} else {
			item.Success()
		}
		batchOpResult = append(batchOpResult, item)
	}

	return batchOpResult, nil
}

// Upgrade batch upgrade
func (b *BatchOperationHandler) Upgrade(ctx context.Context, tenant *dbmodel.Tenants, operator string, batchOpReqs model.BatchOpRequesters) (model.BatchOpResult, error) {
	if logrus.IsLevelEnabled(logrus.DebugLevel) {
		defer util.Elapsed("[BatchOperationHandler] upgrade components")()
	}

	// setup start sequence config
	componentIDs := batchOpReqs.ComponentIDs()
	startupSeqConfigs := b.serviceStartupSequence(componentIDs)

	// chekc allocatable memory
	allocm, err := NewAllocMemory(ctx, b.statusCli, tenant, batchOpReqs)
	if err != nil {
		return nil, errors.WithMessage(err, "new alloc memory")
	}
	batchOpResult := allocm.BatchOpResult()
	validUpgrades := allocm.BatchOpRequests()

	validUpgrades, batchOpResult2 := b.checkEvents(validUpgrades)
	batchOpResult = append(batchOpResult, batchOpResult2...)

	// create events
	if err := b.createEvents(tenant.UUID, operator, batchOpReqs, allocm.BadOpRequests(), allocm.memoryType); err != nil {
		return nil, err
	}

	for _, upgrade := range validUpgrades {
		upgrade.UpdateConfig("boot_seq_dep_service_ids", strings.Join(startupSeqConfigs[upgrade.GetComponentID()], ","))
		err := retryutil.Retry(1*time.Microsecond, 1, func() (bool, error) {
			if err := b.operationHandler.upgrade(upgrade); err != nil {
				return false, err
			}
			return true, nil
		})
		item := upgrade.BatchOpFailureItem()
		if err != nil {
			item.ErrMsg = err.Error()
		} else {
			item.Success()
		}
		batchOpResult = append(batchOpResult, item)
	}
	return batchOpResult, nil
}

func (b *BatchOperationHandler) checkEvents(batchOpReqs model.BatchOpRequesters) (model.BatchOpRequesters, model.BatchOpResult) {
	if logrus.IsLevelEnabled(logrus.DebugLevel) {
		defer util.Elapsed("[BatchOperationHandler] check events")()
	}

	var validReqs model.BatchOpRequesters
	var batchOpResult model.BatchOpResult
	for _, req := range batchOpReqs {
		req := req
		if apiutil.CanDoEvent("", dbmodel.SYNEVENTTYPE, "service", req.GetComponentID(), "") {
			validReqs = append(validReqs, req)
			continue
		}
		item := req.BatchOpFailureItem()
		item.ErrMsg = "The last event has not been completed"
		batchOpResult = append(batchOpResult, item)
	}
	return validReqs, batchOpResult
}

func (b *BatchOperationHandler) createEvents(tenantID, operator string, batchOpReqs, badOpReqs model.BatchOpRequesters, memoryType string) error {
	if logrus.IsLevelEnabled(logrus.DebugLevel) {
		defer util.Elapsed("[BatchOperationHandler] create events")()
	}

	bads := make(map[string]struct{})
	for _, req := range badOpReqs {
		bads[req.GetEventID()] = struct{}{}
	}
	var eventIDs []string
	var events []*dbmodel.ServiceEvent
	for _, req := range batchOpReqs {
		event := &dbmodel.ServiceEvent{
			EventID:    req.GetEventID(),
			TenantID:   tenantID,
			Target:     dbmodel.TargetTypeService,
			TargetID:   req.GetComponentID(),
			UserName:   operator,
			StartTime:  time.Now().Format(time.RFC3339),
			CreateTime: time.Now().Format(time.RFC3339),
			SynType:    dbmodel.ASYNEVENTTYPE,
			OptType:    req.OpType(),
		}
		_, ok := bads[req.GetEventID()]
		if ok {
			event.Reason = memoryType
			event.EndTime = event.StartTime
			event.FinalStatus = "complete"
			event.Status = "failure"

		}
		eventIDs = append(eventIDs, event.EventID)
		events = append(events, event)
	}
	b.EventIDs = eventIDs
	return db.GetManager().DB().Transaction(func(tx *gorm.DB) error {
		return db.GetManager().ServiceEventDaoTransactions(tx).CreateEventsInBatch(events)
	})
}

// ServiceDependency documents a set of services and their dependencies.
// provides the ability to build linked lists of dependencies and find circular dependencies.
type ServiceDependency struct {
	serviceIDs  []string
	sid2depsids map[string][]string
	depsid2sids map[string][]string
}

// NewServiceDependency creates a new ServiceDependency.
func NewServiceDependency(serviceIDs []string) (*ServiceDependency, error) {
	relations, err := db.GetManager().TenantServiceRelationDao().ListByServiceIDs(serviceIDs)
	if err != nil {
		return nil, fmt.Errorf("list retions: %v", err)
	}
	sid2depsids := make(map[string][]string)
	depsid2sids := make(map[string][]string)
	for _, relation := range relations {
		sid2depsids[relation.ServiceID] = append(sid2depsids[relation.ServiceID], relation.DependServiceID)
		depsid2sids[relation.DependServiceID] = append(depsid2sids[relation.DependServiceID], relation.ServiceID)
	}

	logrus.Debugf("create a new ServiceDependency; sid2depsids: %#v; depsid2sids: %#v", sid2depsids, depsid2sids)
	return &ServiceDependency{
		serviceIDs:  serviceIDs,
		sid2depsids: sid2depsids,
		depsid2sids: depsid2sids,
	}, nil
}

// The order in which services are started is determined by their dependencies. If interdependencies occur, one of them is ignored.
func (s *ServiceDependency) serviceStartupSequence() map[string][]string {
	headNodes := s.headNodes()
	var lists []*list.List
	for _, h := range headNodes {
		l := list.New()
		l.PushBack(h)
		lists = append(lists, s.buildLinkListByHead(l)...)
	}

	result := make(map[string][]string)
	for _, l := range lists {
		cur := l.Front()
		for cur != nil && cur.Next() != nil {
			existingVals := result[cur.Value.(string)]
			exists := false
			for _, val := range existingVals {
				if val == cur.Next().Value.(string) {
					exists = true
					break
				}
			}
			if !exists {
				result[cur.Value.(string)] = append(result[cur.Value.(string)], cur.Next().Value.(string))
			}
			cur = cur.Next()
		}
	}

	return result
}

// headNodes finds out the service ID of all head nodes. The head nodes are services that are not dependent on other services.
func (s *ServiceDependency) headNodes() []string {
	var headNodes []string
	for _, sid := range s.serviceIDs {
		if _, ok := s.depsid2sids[sid]; ok {
			continue
		}

		headNodes = append(headNodes, sid)
	}

	// if there is no head node(i.e. a->b->c->d->a), then a node is randomly selected.
	// however, this node cannot be a tail node
	for _, sid := range s.serviceIDs {
		// does not depend on other services, it is the tail node
		if _, ok := s.sid2depsids[sid]; !ok {
			continue
		}

		headNodes = append(headNodes, sid)
		logrus.Debugf("randomly select '%s' as the head node", sid)
		break
	}

	return headNodes
}

// buildLinkListByHead recursively creates linked lists of service dependencies.
//
// recursive end condition:
// 1. nil or empty input
// 2. no more children
// 3. child node is already in the linked list
func (s *ServiceDependency) buildLinkListByHead(l *list.List) []*list.List {
	// nil or empty input
	if l == nil || l.Len() == 0 {
		return nil
	}

	// the last node is the head node of the new linked list
	sid, _ := l.Back().Value.(string)
	depsids, ok := s.sid2depsids[sid]
	// no more children
	if !ok {
		copy := list.New()
		copy.PushBackList(l)
		return []*list.List{copy}
	}

	var result []*list.List
	for _, depsid := range depsids {
		// child node is already in the linked list
		if alreadyInLinkedList(l, depsid) || s.childInLinkedList(l, depsid) {
			copy := list.New()
			copy.PushBackList(l)
			result = append(result, copy)
			continue
		}

		newl := list.New()
		newl.PushBackList(l)
		newl.PushBack(depsid)

		sublists := s.buildLinkListByHead(newl)
		if len(sublists) == 0 {
			result = append(result, newl)
		} else {
			result = append(result, sublists...)
		}
	}

	return result
}

func (s *ServiceDependency) childInLinkedList(l *list.List, sid string) bool {
	depsids, ok := s.sid2depsids[sid]
	if !ok {
		return false
	}

	for _, depsid := range depsids {
		if alreadyInLinkedList(l, depsid) {
			return true
		}
	}

	return false
}

func alreadyInLinkedList(l *list.List, depsid string) bool {
	pre := l.Back()
	for pre != nil {
		val := pre.Value.(string)
		if val == depsid {
			return true
		}
		pre = pre.Prev()
	}

	return false
}

// AllocMemory represents a allocatable memory.
type AllocMemory struct {
	tenant          *dbmodel.Tenants
	allcm           *int64
	memoryType      string
	components      map[string]*dbmodel.TenantServices
	batchOpResult   model.BatchOpResult
	batchOpRequests model.BatchOpRequesters
	badOpRequest    model.BatchOpRequesters
}

// NewAllocMemory creates a new AllocMemory.
func NewAllocMemory(ctx context.Context, statusCli *client.AppRuntimeSyncClient, tenant *dbmodel.Tenants, batchOpReqs model.BatchOpRequesters) (*AllocMemory, error) {
	if logrus.IsLevelEnabled(logrus.DebugLevel) {
		defer util.Elapsed("[NewAllocMemory] check allocatable memory")()
	}

	am := &AllocMemory{
		tenant: tenant,
	}

	if tenant.LimitMemory != 0 {
		tenantUsedResource, err := statusCli.GetTenantResource(tenant.UUID)
		if err != nil {
			return nil, err
		}
		allocm := tenant.LimitMemory - int(tenantUsedResource.MemoryLimit)
		am.allcm = util.Int64(int64(allocm))
		am.memoryType = "tenant_lack_of_memory"
	} else {
		allcm, err := ClusterAllocMemory(ctx)
		if err != nil {
			return nil, err
		}
		am.allcm = util.Int64(allcm)
		am.memoryType = "cluster_lack_of_memory"
	}

	components, err := am.listComponents(batchOpReqs.ComponentIDs())
	if err != nil {
		return nil, err
	}
	am.components = components

	// check alloc memory for every components.
	var reqs model.BatchOpRequesters
	var batchOpResult model.BatchOpResult
	var badOpRequest model.BatchOpRequesters
	for _, req := range batchOpReqs {
		req := req
		if err := am.check(req.GetComponentID()); err != nil {
			item := req.BatchOpFailureItem()
			item.ErrMsg = err.Error()
			batchOpResult = append(batchOpResult, item)
			badOpRequest = append(badOpRequest, req)
			continue
		}
		reqs = append(reqs, req)
	}
	am.batchOpResult = batchOpResult
	am.batchOpRequests = reqs
	am.badOpRequest = badOpRequest

	return am, nil
}

// BatchOpResult returns the batchOpResult.
func (a *AllocMemory) BatchOpResult() model.BatchOpResult {
	return a.batchOpResult
}

// BatchOpRequests returns the batchOpRequests.
func (a *AllocMemory) BatchOpRequests() model.BatchOpRequesters {
	return a.batchOpRequests
}

// BadOpRequests returns the badOpRequests.
func (a *AllocMemory) BadOpRequests() model.BatchOpRequesters {
	return a.badOpRequest
}

func (a *AllocMemory) listComponents(componentIDs []string) (map[string]*dbmodel.TenantServices, error) {
	components, err := db.GetManager().TenantServiceDao().GetServiceByIDs(componentIDs)
	if err != nil {
		return nil, err
	}

	// make a map for compoenents
	res := make(map[string]*dbmodel.TenantServices)
	for _, cpt := range components {
		cpt := cpt
		res[cpt.ServiceID] = cpt
	}
	return res, nil
}

func (a *AllocMemory) check(componentID string) error {
	component, ok := a.components[componentID]
	if !ok {
		return errors.New("component not found")
	}
	requestMemory := component.ContainerMemory * component.Replicas

	allom := util.Int64Value(a.allcm)
	if requestMemory > int(allom) {
		logrus.Errorf("request memory is %d, but got %d allocatable memory", requestMemory, allom)
		return errors.New("tenant_lack_of_memory")
	}

	*a.allcm -= int64(requestMemory)

	return nil
}