Merge pull request #810 in YUNIONIO/onecloud from ~QIUJIAN/onecloud:hotfix/qj-fix-db-rows-not-close to release/2.4.0

* commit 'fdaec75067e0ddd2792546d0c4506fcd7cf372bd':
  在rowsToArray中进行defer rows.Close()
  修正：1. sql.Rows使用后未关闭，导致占用db链接 2. 服务启动后，调用AppDBInit方法设置db连接池上限，避免耗尽mysql连接 3. app和workmanager增加db属性，用来标示该worker和app是否使用数据库，从而估计需要设置的数据库连接的上限

pkg/appsrv/appsrv.go

@@ -47,12 +47,12 @@ const (
 	DEFAULT_PROCESS_TIMEOUT     = 15 * time.Second
 )
 
-func NewApplication(name string, connMax int) *Application {
+func NewApplication(name string, connMax int, db bool) *Application {
 	app := Application{name: name,
 		context:           context.Background(),
 		connMax:           connMax,
-		session:           NewWorkerManager("HttpRequestWorkerManager", connMax, DEFAULT_BACKLOG),
-		systemSession:     NewWorkerManager("InternalHttpRequestWorkerManager", 1, DEFAULT_BACKLOG),
+		session:           NewWorkerManager("HttpRequestWorkerManager", connMax, DEFAULT_BACKLOG, db),
+		systemSession:     NewWorkerManager("InternalHttpRequestWorkerManager", 1, DEFAULT_BACKLOG, false),
 		roots:             make(map[string]*RadixNode),
 		rootLock:          &sync.RWMutex{},
 		idleTimeout:       DEFAULT_IDLE_TIMEOUT,
@@ -273,7 +273,7 @@ func (app *Application) defaultHandle(w http.ResponseWriter, r *http.Request, ri
 	return nil, nil
 }
 
-func (app *Application) addDefaultHandler(method string, prefix string, handler func(context.Context, http.ResponseWriter, *http.Request), name string) {
+func (app *Application) AddDefaultHandler(method string, prefix string, handler func(context.Context, http.ResponseWriter, *http.Request), name string) {
 	segs := SplitPath(prefix)
 	hi := newHandlerInfo(method, segs, handler, nil, name, nil)
 	hi.SetSkipLog(true).SetWorkerManager(app.systemSession)
@@ -281,11 +281,11 @@ func (app *Application) addDefaultHandler(method string, prefix string, handler
 }
 
 func (app *Application) addDefaultHandlers() {
-	app.addDefaultHandler("GET", "/version", VersionHandler, "version")
-	app.addDefaultHandler("GET", "/stats", StatisticHandler, "stats")
-	app.addDefaultHandler("POST", "/ping", PingHandler, "ping")
-	app.addDefaultHandler("GET", "/ping", PingHandler, "ping")
-	app.addDefaultHandler("GET", "/worker_stats", WorkerStatsHandler, "worker_stats")
+	app.AddDefaultHandler("GET", "/version", VersionHandler, "version")
+	app.AddDefaultHandler("GET", "/stats", StatisticHandler, "stats")
+	app.AddDefaultHandler("POST", "/ping", PingHandler, "ping")
+	app.AddDefaultHandler("GET", "/ping", PingHandler, "ping")
+	app.AddDefaultHandler("GET", "/worker_stats", WorkerStatsHandler, "worker_stats")
 }
 
 func timeoutHandle(h http.Handler) http.HandlerFunc {
@@ -301,11 +301,12 @@ func timeoutHandle(h http.Handler) http.HandlerFunc {
 }
 
 func (app *Application) initServer(addr string) *http.Server {
-	db := AppContextDB(app.context)
+	/* db := AppContextDB(app.context)
 	if db != nil {
 		db.SetMaxIdleConns(app.connMax + 1)
 		db.SetMaxOpenConns(app.connMax + 1)
 	}
+	*/
 	app.addDefaultHandlers()
 	s := &http.Server{
 		Addr:              addr,

pkg/appsrv/workers.go

@@ -129,9 +129,10 @@ type SWorkerManager struct {
 	detachedWorker *SWorkerList
 	workerLock     *sync.Mutex
 	workerId       uint64
+	dbWorker       bool
 }
 
-func NewWorkerManager(name string, workerCount int, backlog int) *SWorkerManager {
+func NewWorkerManager(name string, workerCount int, backlog int, dbWorker bool) *SWorkerManager {
 	manager := SWorkerManager{name: name,
 		queue:          NewRing(workerCount * backlog),
 		workerCount:    workerCount,
@@ -139,7 +140,9 @@ func NewWorkerManager(name string, workerCount int, backlog int) *SWorkerManager
 		activeWorker:   newWorkerList(),
 		detachedWorker: newWorkerList(),
 		workerLock:     &sync.Mutex{},
-		workerId:       0}
+		workerId:       0,
+		dbWorker:       dbWorker,
+	}
 
 	workerManagers = append(workerManagers, &manager)
 	return &manager
@@ -206,7 +209,7 @@ func (wm *SWorkerManager) scheduleWithLock() {
 		}
 		go worker.run()
 	} else {
-		log.Warningf("[%s] BUSY activeWork %d max %d queue: %d", wm, wm.activeWorker.size(), wm.workerCount, wm.queue.Size())
+		log.Warningf("[%s] BUSY activeWork %d detachedWork %d max %d queue: %d", wm, wm.ActiveWorkerCount(), wm.DetachedWorkerCount(), wm.workerCount, wm.queue.Size())
 	}
 }
 
@@ -247,3 +250,13 @@ func WorkerStatsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	result.Add(jsonutils.Marshal(&stats), "workers")
 	fmt.Fprintf(w, result.String())
 }
+
+func GetDBConnectionCount() int {
+	conn := 0
+	for i := 0; i < len(workerManagers); i += 1 {
+		if workerManagers[i].dbWorker {
+			conn += workerManagers[i].workerCount
+		}
+	}
+	return conn
+}

pkg/cloudcommon/app.go

@@ -8,9 +8,9 @@ import (
 	"yunion.io/x/onecloud/pkg/appsrv"
 )
 
-func InitApp(options *Options) *appsrv.Application {
+func InitApp(options *Options, dbAccess bool) *appsrv.Application {
 	// cache := appsrv.NewCache(options.AuthTokenCacheSize)
-	app := appsrv.NewApplication(options.ApplicationID, options.RequestWorkerCount)
+	app := appsrv.NewApplication(options.ApplicationID, options.RequestWorkerCount, dbAccess)
 	app.CORSAllowHosts(options.CorsHosts)
 
 	// app.SetContext(appsrv.APP_CONTEXT_KEY_CACHE, cache)
@@ -21,6 +21,8 @@ func InitApp(options *Options) *appsrv.Application {
 }
 
 func ServeForever(app *appsrv.Application, options *Options) {
+	AppDBInit(app)
+
 	addr := net.JoinHostPort(options.Address, strconv.Itoa(options.Port))
 	proto := "http"
 	if options.EnableSsl {

pkg/cloudcommon/cronman/cronman.go

@@ -8,6 +8,7 @@ import (
 
 	"yunion.io/x/log"
 	"yunion.io/x/onecloud/pkg/appctx"
+	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
 )
@@ -18,7 +19,8 @@ var manager *SCronJobManager
 
 func init() {
 	manager = &SCronJobManager{
-		jobs: make([]*SCronJob, 0),
+		jobs:    make([]*SCronJob, 0),
+		workers: appsrv.NewWorkerManager("CronJobWorkers", 4, 1024, true),
 	}
 }
 
@@ -88,6 +90,7 @@ type SCronJobManager struct {
 	add     chan *SCronJob
 	stop    chan struct{}
 	running bool
+	workers *appsrv.SWorkerManager
 }
 
 func GetCronJobManager() *SCronJobManager {
@@ -171,7 +174,7 @@ func (self *SCronJobManager) run() {
 				if job.Next.After(now) || job.Next.IsZero() {
 					break
 				}
-				go job.runJob(false)
+				job.runJob(false)
 				job.Next = job.Timer.Next(now)
 				heap.Fix(&self.jobs, i)
 			}
@@ -187,6 +190,12 @@ func (self *SCronJobManager) run() {
 }
 
 func (job *SCronJob) runJob(isStart bool) {
+	manager.workers.Run(func() {
+		job.runJobInWorker(isStart)
+	}, nil, nil)
+}
+
+func (job *SCronJob) runJobInWorker(isStart bool) {
 	defer func() {
 		if r := recover(); r != nil {
 			log.Errorf("CronJob task %s run error: %s", job.Name, r)

pkg/cloudcommon/database.go

@@ -1,15 +1,24 @@
 package cloudcommon
 
 import (
+	"context"
 	"database/sql"
+	"fmt"
+	"net/http"
 
+	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
 	"yunion.io/x/sqlchemy"
 
+	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
 )
 
+const (
+	MIN_DB_CONN_MAX = 5
+)
+
 func InitDB(options *DBOptions) {
 	if options.GlobalVirtualResourceNamespace {
 		consts.EnableGlobalVirtualResourceNamespace()
@@ -33,3 +42,28 @@ func InitDB(options *DBOptions) {
 func CloseDB() {
 	sqlchemy.CloseDB()
 }
+
+func AppDBInit(app *appsrv.Application) {
+	dbConn := sqlchemy.GetDB()
+	if dbConn != nil {
+		connMax := appsrv.GetDBConnectionCount()
+		if connMax < MIN_DB_CONN_MAX {
+			connMax = MIN_DB_CONN_MAX
+		}
+		log.Infof("Total %d db workers, set db connection max", connMax)
+		dbConn.SetMaxIdleConns(connMax)
+		dbConn.SetMaxOpenConns(connMax*2 + 1)
+	}
+
+	app.AddDefaultHandler("GET", "/db_stats", DBStatsHandler, "db_stats")
+}
+
+func DBStatsHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	result := jsonutils.NewDict()
+	dbConn := sqlchemy.GetDB()
+	if dbConn != nil {
+		stats := dbConn.Stats()
+		result.Add(jsonutils.Marshal(&stats), "db_stats")
+	}
+	fmt.Fprintf(w, result.String())
+}

pkg/cloudcommon/db/db_dispatcher.go

@@ -305,10 +305,6 @@ func listItemQueryFilters(manager IModelManager, ctx context.Context, q *sqlchem
 }
 
 func query2List(manager IModelManager, ctx context.Context, userCred mcclient.TokenCredential, q *sqlchemy.SQuery, query jsonutils.JSONObject) ([]jsonutils.JSONObject, error) {
-	rows, err := q.Rows()
-	if err != nil {
-		return nil, err
-	}
 	listF := listFields(manager, userCred)
 	fieldFilter := jsonutils.GetQueryStringArray(query, "field")
 	if len(fieldFilter) > 0 && IsAdminAllowList(userCred, manager) {
@@ -332,6 +328,11 @@ func query2List(manager IModelManager, ctx context.Context, userCred mcclient.To
 		return nil, err
 	}
 	results := make([]jsonutils.JSONObject, 0)
+	rows, err := q.Rows()
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
 	for rows.Next() {
 		itemValue := reflect.Indirect(reflect.ValueOf(item))
 		itemValue.Set(itemInitValue)
@@ -758,6 +759,7 @@ func FetchModelObjects(modelManager IModelManager, query *sqlchemy.SQuery, targe
 		}
 		return err
 	}
+	defer rows.Close()
 
 	targetsValue := reflect.Indirect(reflect.ValueOf(targets))
 	for rows.Next() {

pkg/cloudcommon/db/taskman/handler.go

@@ -11,7 +11,7 @@ import (
 var taskWorkMan *appsrv.SWorkerManager
 
 func init() {
-	taskWorkMan = appsrv.NewWorkerManager("TaskWorkerManager", 4, 1024)
+	taskWorkMan = appsrv.NewWorkerManager("TaskWorkerManager", 4, 1024, true)
 }
 
 func AddTaskHandler(prefix string, app *appsrv.Application) {

pkg/cloudcommon/db/taskman/localtaskworker.go

@@ -12,7 +12,7 @@ import (
 var localTaskWorkerMan *appsrv.SWorkerManager
 
 func init() {
-	localTaskWorkerMan = appsrv.NewWorkerManager("LocalTaskWorkerManager", 4, 1024)
+	localTaskWorkerMan = appsrv.NewWorkerManager("LocalTaskWorkerManager", 4, 1024, false)
 }
 
 func Error2TaskData(err error) jsonutils.JSONObject {

pkg/cloudcommon/db/taskman/taskobjs.go

@@ -1,6 +1,8 @@
 package taskman
 
 import (
+	"database/sql"
+
 	"yunion.io/x/log"
 
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
@@ -29,9 +31,12 @@ func (manager *STaskObjectManager) GetObjectIds(task *STask) []string {
 	q := taskobjs.Query(taskobjs.Field("obj_id")).Equals("task_id", task.Id)
 	rows, err := q.Rows()
 	if err != nil {
-		log.Errorf("TaskObjectManager GetObjectIds fail %s", err)
+		if err != sql.ErrNoRows {
+			log.Errorf("TaskObjectManager GetObjectIds fail %s", err)
+		}
 		return nil
 	}
+	defer rows.Close()
 	for rows.Next() {
 		var objId string
 		err = rows.Scan(&objId)

pkg/compute/models/capabilities.go

@@ -57,6 +57,7 @@ func getHypervisors(zone *SZone) []string {
 	if err != nil {
 		return nil
 	}
+	defer rows.Close()
 	hypervisors := make([]string, 0)
 	for rows.Next() {
 		var hostType string
@@ -80,6 +81,7 @@ func getStorageTypes(zone *SZone) []string {
 	if err != nil {
 		return nil
 	}
+	defer rows.Close()
 	storageTypes := make([]string, 0)
 	for rows.Next() {
 		var storageType, mediumType string
@@ -106,6 +108,7 @@ func getGPUs(zone *SZone) []string {
 	if err != nil {
 		return nil
 	}
+	defer rows.Close()
 	gpus := make([]string, 0)
 	for rows.Next() {
 		var model string

pkg/compute/models/cloudaccounts.go

@@ -588,6 +588,7 @@ func (self *SCloudaccount) getProjectIds() []string {
 	if err != nil {
 		return nil
 	}
+	defer rows.Close()
 	ret := make([]string, 0)
 	for rows.Next() {
 		var projId string
@@ -606,6 +607,7 @@ func (self *SCloudaccount) getVersion() string {
 	if err != nil {
 		return ""
 	}
+	defer rows.Close()
 	ret := make([]string, 0)
 	for rows.Next() {
 		var projId string

pkg/compute/models/guestnetworks.go

@@ -574,6 +574,7 @@ func (manager *SGuestnetworkManager) getRecentlyReleasedIPAddresses(networkId st
 		log.Errorf("GetRecentlyReleasedIPAddresses fail %s", err)
 		return nil
 	}
+	defer rows.Close()
 	ret := make(map[string]bool)
 	for rows.Next() {
 		var ip string

pkg/compute/models/guests.go

@@ -3330,6 +3330,7 @@ func (manager *SGuestManager) GetIpInProjectWithName(projectId, name string, isE
 		log.Errorf("Get guest ip with name query err: %v", err)
 		return ips
 	}
+	defer rows.Close()
 	for rows.Next() {
 		var ip string
 		err = rows.Scan(&ip)

pkg/compute/models/hosts.go

@@ -3494,6 +3494,7 @@ func (manager *SHostManager) GetEsxiAgentHostId(key string) (string, error) {
 	if err != nil {
 		return "", err
 	}
+	defer rows.Close()
 
 	var hostId string
 	hostIds := make([]string, 0)

pkg/compute/models/loadbalanceragents.go

@@ -330,6 +330,7 @@ func (man *SLoadbalancerAgentManager) CleanPendingDeleteLoadbalancers(ctx contex
 				log.Errorf("%s: query pending_deleted_at < %s: %s", keyPlural, minT, err)
 				continue
 			}
+			defer rows.Close()
 			m, err := db.NewModelObject(man)
 			if err != nil {
 				log.Errorf("%s: new model object failed: %s", keyPlural, err)

pkg/compute/models/networks.go

@@ -189,6 +189,7 @@ func (self *SNetwork) GetUsedAddresses() map[string]bool {
 			log.Errorf("GetUsedAddresses query fail: %s", err)
 			return nil
 		}
+		defer rows.Close()
 		for rows.Next() {
 			var ip string
 			err = rows.Scan(&ip)
@@ -298,6 +299,7 @@ func (self *SNetwork) GetUsedIfnames() map[string]bool {
 		log.Errorf("GetUsedIfnames query fail: %s", err)
 		return nil
 	}
+	defer rows.Close()
 	for rows.Next() {
 		var ifname string
 		err = rows.Scan(&ifname)

pkg/compute/models/specs.go

@@ -92,6 +92,7 @@ func ListItems(manager db.IModelManager, ctx context.Context, userCred mcclient.
 	if err != nil {
 		return nil, err
 	}
+	defer rows.Close()
 	items := make([]ISpecModel, 0)
 	for rows.Next() {
 		item, err := db.NewModelObject(manager)

pkg/compute/service/service.go

@@ -48,7 +48,7 @@ func StartService() {
 	cloudcommon.InitDB(&options.Options.DBOptions)
 	defer cloudcommon.CloseDB()
 
-	app := cloudcommon.InitApp(&options.Options.Options)
+	app := cloudcommon.InitApp(&options.Options.Options, true)
 
 	compute.InitHandlers(app)
 

pkg/scheduler/db/models/base.go

@@ -136,6 +136,8 @@ func virtualResourceRowsNotDeletedIn(r Resourcer, key string, set []string) (*sq
 }
 
 func rowsToArray(r Resourcer, rows *sql.Rows) ([]interface{}, error) {
+	defer rows.Close()
+
 	columns, _ := rows.Columns()
 
 	objs := make([]interface{}, 0, len(columns))

pkg/util/logclient/logclient.go

@@ -70,7 +70,7 @@ var BLACK_LIST_OBJ_TYPE = []string{"parameter"}
 var logclientWorkerMan *appsrv.SWorkerManager
 
 func init() {
-	logclientWorkerMan = appsrv.NewWorkerManager("LogClientWorkerManager", 1, 50)
+	logclientWorkerMan = appsrv.NewWorkerManager("LogClientWorkerManager", 1, 50, false)
 }
 
 type IObject interface {

pkg/webconsole/service/service.go

@@ -48,7 +48,7 @@ func StartService() {
 }
 
 func start() {
-	app := cloudcommon.InitApp(&o.Options.Options)
+	app := cloudcommon.InitApp(&o.Options.Options, false)
 	webconsole.InitHandlers(app)
 
 	root := mux.NewRouter()

pkg/yunionconf/service/service.go

@@ -30,7 +30,7 @@ func StartService() {
 	cloudcommon.InitDB(&options.Options.DBOptions)
 	defer cloudcommon.CloseDB()
 
-	app := cloudcommon.InitApp(&options.Options.Options)
+	app := cloudcommon.InitApp(&options.Options.Options, true)
 	yunionconf.InitHandlers(app)
 
 	if db.CheckSync(options.Options.AutoSyncTable) {