Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat:opt resource manager, issue: #332 #333

Merged
merged 3 commits into from
Dec 19, 2024
Merged

feat:opt resource manager, issue: #332 #333

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f099491
feat:opt resource manager, issue: #332
tbs60 Dec 19, 2024
69b954c
feat:opt resource manager, issue: #332
flyy1012 Dec 19, 2024
13690bd
feat:opt resource manager, issue: #332
tbs60 Dec 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 59 additions & 56 deletions src/backend/booster/server/pkg/resource/crm/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,8 +96,8 @@ type ResourceManager interface {
// RegisterUser receive a user and return a HandlerWithUser which operates under this user.
RegisterUser(user string) (HandlerWithUser, error)

// GetResourceDetail return details of resource description.
GetResourceDetail() *rsc.Details
// // GetResourceDetail return details of resource description.
// GetResourceDetail() *rsc.Details

// Run the manager.
Run() error
Expand Down Expand Up @@ -197,7 +197,7 @@ type resourceManager struct {

// following should be init in recover
registeredResourceMap map[string]*resource
registeredResourceMapLock sync.Mutex
registeredResourceMapLock sync.RWMutex

nodeInfoPool *op.NodeInfoPool

Expand Down Expand Up @@ -227,13 +227,13 @@ func (rm *resourceManager) RegisterUser(user string) (HandlerWithUser, error) {
return rm.registerUser(user)
}

// GetResourceDetail get the details of all resources in pool.
func (rm *resourceManager) GetResourceDetail() *rsc.Details {
return &rsc.Details{
Rsc: rm.rscDetail,
App: rm.appDetail,
}
}
// // GetResourceDetail get the details of all resources in pool.
// func (rm *resourceManager) GetResourceDetail() *rsc.Details {
// return &rsc.Details{
// Rsc: rm.rscDetail,
// App: rm.appDetail,
// }
// }

// Run the resource manager. Listen to role change events. Start manager when master, stop when not.
func (rm *resourceManager) Run() error {
Expand Down Expand Up @@ -279,7 +279,7 @@ func (rm *resourceManager) start() {
go rm.runLockCleaner()
go rm.runBrokerChecker()
go rm.runRscDetailSync()
go rm.runAppDetailSync()
// go rm.runAppDetailSync()
}

func (rm *resourceManager) stop() {
Expand Down Expand Up @@ -370,47 +370,47 @@ func (rm *resourceManager) runRscDetailSync() {
}
}

func (rm *resourceManager) runAppDetailSync() {
blog.Infof("crm: begin to run app detail sync")
ticker := time.NewTimer(syncAppDetailTimeGap)
defer ticker.Stop()

for {
select {
case <-rm.ctx.Done():
blog.Warnf("crm: app detail sync done")
return
case <-ticker.C:
if rm.registeredResourceMap == nil {
continue
}

appDetail := make([]*rsc.AppDetails, 0, 100)
rm.registeredResourceMapLock.Lock()
for _, r := range rm.registeredResourceMap {
if r.status == resourceStatusReleased {
continue
}

appDetail = append(appDetail, &rsc.AppDetails{
ResourceID: r.resourceID,
BrokerID: r.brokerResourceID,
BrokerName: r.brokerName,
BrokerSold: r.brokerSold,
User: r.user,
Status: r.status.String(),
Image: r.param.Image,
CreateTime: r.initTime,
RequestInstance: r.requestInstance,
NotReadyInstance: r.noReadyInstance,
Label: r.param.City,
})
}
rm.registeredResourceMapLock.Unlock()
rm.appDetail = appDetail
}
}
}
// func (rm *resourceManager) runAppDetailSync() {
// blog.Infof("crm: begin to run app detail sync")
// ticker := time.NewTimer(syncAppDetailTimeGap)
// defer ticker.Stop()

// for {
// select {
// case <-rm.ctx.Done():
// blog.Warnf("crm: app detail sync done")
// return
// case <-ticker.C:
// if rm.registeredResourceMap == nil {
// continue
// }

// appDetail := make([]*rsc.AppDetails, 0, 100)
// rm.registeredResourceMapLock.RLock()
// for _, r := range rm.registeredResourceMap {
// if r.status == resourceStatusReleased {
// continue
// }

// appDetail = append(appDetail, &rsc.AppDetails{
// ResourceID: r.resourceID,
// BrokerID: r.brokerResourceID,
// BrokerName: r.brokerName,
// BrokerSold: r.brokerSold,
// User: r.user,
// Status: r.status.String(),
// Image: r.param.Image,
// CreateTime: r.initTime,
// RequestInstance: r.requestInstance,
// NotReadyInstance: r.noReadyInstance,
// Label: r.param.City,
// })
// }
// rm.registeredResourceMapLock.RUnlock()
// rm.appDetail = appDetail
// }
// }
// }

func (rm *resourceManager) logResourceStats() {
blog.Infof("crm: report resources(%s) following: %s", rm.conf.Operator, rm.nodeInfoPool.GetStats())
Expand Down Expand Up @@ -679,14 +679,16 @@ func (rm *resourceManager) createResources(r *resource) error {

func (rm *resourceManager) saveResources(r *resource) error {
rm.registeredResourceMapLock.Lock()
defer rm.registeredResourceMapLock.Unlock()
// defer rm.registeredResourceMapLock.Unlock()

rm.registeredResourceMap[r.resourceID] = r
if r.status == resourceStatusReleased {
delete(rm.registeredResourceMap, r.resourceID)
}
tr := resource2Table(r)
rm.registeredResourceMapLock.Unlock()

return rm.mysql.PutResource(resource2Table(r))
return rm.mysql.PutResource(tr)
}

func (rm *resourceManager) updateResourcesCache(r *resource) error {
Expand All @@ -699,8 +701,8 @@ func (rm *resourceManager) updateResourcesCache(r *resource) error {
}

func (rm *resourceManager) getResources(resourceID string) (*resource, error) {
rm.registeredResourceMapLock.Lock()
defer rm.registeredResourceMapLock.Unlock()
rm.registeredResourceMapLock.RLock()
defer rm.registeredResourceMapLock.RUnlock()

r, ok := rm.registeredResourceMap[resourceID]
if !ok {
Expand Down Expand Up @@ -858,6 +860,7 @@ func (rm *resourceManager) launch(
blog.Errorf("crm: try launching service failed, resource(%s) user(%s): %v", resourceID, user, err)
return err
}
blog.Infof("crm: try launching service, resource(%s) for user(%s)", resourceID, user)

// specify city
originCity := r.param.City
Expand Down
14 changes: 14 additions & 0 deletions src/backend/booster/server/pkg/resource/crm/tables.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,15 @@ import (

"github.com/TencentBlueKing/bk-turbo/src/backend/booster/common/blog"
commonMySQL "github.com/TencentBlueKing/bk-turbo/src/backend/booster/common/mysql"
selfMetric "github.com/TencentBlueKing/bk-turbo/src/backend/booster/server/pkg/metric"

"github.com/jinzhu/gorm"
)

const (
MetricPrefix = "crm"
)

// Basic define the basic table columns.
type Basic struct {
UpdatedAt time.Time `gorm:"column:update_at" json:"-"`
Expand Down Expand Up @@ -134,6 +139,7 @@ func (m *mysql) ensureTables(tables ...interface{}) error {
// ListResource list resource from db, return list and total num.
// list cut by offset and limit, but total num describe the true num.
func (m *mysql) ListResource(opts commonMySQL.ListOptions) ([]*TableResource, int64, error) {
defer timeMetricRecord("list_resource")()
defer logSlowFunc(time.Now().Unix(), "ListResource", 2)
var rl []*TableResource
db := opts.AddWhere(m.db.Model(&TableResource{}).Table(m.conf.MySQLTable))
Expand All @@ -158,6 +164,7 @@ func (m *mysql) ListResource(opts commonMySQL.ListOptions) ([]*TableResource, in

// GetResource get resource.
func (m *mysql) GetResource(resourceID string) (*TableResource, error) {
defer timeMetricRecord("get_resource")()
defer logSlowFunc(time.Now().Unix(), "GetResource", 2)
opts := commonMySQL.NewListOptions()
opts.Limit(1)
Expand All @@ -178,6 +185,7 @@ func (m *mysql) GetResource(resourceID string) (*TableResource, error) {

// CreateResource create a new resource into database.
func (m *mysql) CreateResource(tr *TableResource) error {
defer timeMetricRecord("create_resource")()
defer logSlowFunc(time.Now().Unix(), "CreateResource", 2)
blog.Infof("crm: try to create resource(%s)", tr.ResourceID)
tr.tableName = m.conf.MySQLTable
Expand All @@ -191,6 +199,7 @@ func (m *mysql) CreateResource(tr *TableResource) error {

// PutResource update a existing resource with full fields into database.
func (m *mysql) PutResource(tr *TableResource) error {
defer timeMetricRecord("put_resource")()
defer logSlowFunc(time.Now().Unix(), "PutResource", 2)
blog.Infof("crm: try to put resource(%s)", tr.ResourceID)
tr.tableName = m.conf.MySQLTable
Expand All @@ -204,6 +213,7 @@ func (m *mysql) PutResource(tr *TableResource) error {

// DeleteResource delete resource from db. Just set the disabled to true instead of real deletion.
func (m *mysql) DeleteResource(resourceID string) error {
defer timeMetricRecord("delete_resource")()
defer logSlowFunc(time.Now().Unix(), "DeleteResource", 2)
blog.Infof("crm: try to delete resource(%s)", resourceID)
if err := m.db.
Expand All @@ -218,6 +228,10 @@ func (m *mysql) DeleteResource(resourceID string) error {
return nil
}

func timeMetricRecord(operation string) func() {
return selfMetric.TimeMetricRecord(fmt.Sprintf("%s_%s", MetricPrefix, operation))
}

func logSlowFunc(start int64, funcname string, threshold int64) {
now := time.Now().Unix()
if now-start > threshold {
Expand Down
Loading