public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #13400 from connection-reset/concurrent-sql

Browse Source 
Run queries for sql data sources for panels with multiple queries concurrently
Also make max open connections, max idle connections and max connection 
lifetime configurable in the data source settings
This commit is contained in:
Marcus Efraimsson
2018-10-10 19:12:38 +02:00
committed by GitHub
parent 39b25e0596 4757ca2242
commit bc3b7357bd
8 changed files with 152 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tsdb/sql_engine.go
+35 -23
View File
@@ -98,8 +98,12 @@ var NewSqlQueryEndpoint = func(config *SqlQueryEndpointConfiguration, rowTransfo
return nil, err
}
engine.SetMaxOpenConns(10)
engine.SetMaxIdleConns(10)
maxOpenConns := config.Datasource.JsonData.Get("maxOpenConns").MustInt(0)
engine.SetMaxOpenConns(maxOpenConns)
maxIdleConns := config.Datasource.JsonData.Get("maxIdleConns").MustInt(2)
engine.SetMaxIdleConns(maxIdleConns)
connMaxLifetime := config.Datasource.JsonData.Get("connMaxLifetime").MustInt(14400)
engine.SetConnMaxLifetime(time.Duration(connMaxLifetime) * time.Second)
engineCache.versions[config.Datasource.Id] = config.Datasource.Version
engineCache.cache[config.Datasource.Id] = engine
@@ -116,9 +120,7 @@ func (e *sqlQueryEndpoint) Query(ctx context.Context, dsInfo *models.DataSource,
Results: make(map[string]*QueryResult),
}
session := e.engine.NewSession()
defer session.Close()
db := session.DB()
var wg sync.WaitGroup
for _, query := range tsdbQuery.Queries {
rawSQL := query.Model.Get("rawSql").MustString()
@@ -145,31 +147,41 @@ func (e *sqlQueryEndpoint) Query(ctx context.Context, dsInfo *models.DataSource,
queryResult.Meta.Set("sql", rawSQL)
rows, err := db.Query(rawSQL)
if err != nil {
queryResult.Error = err
continue
}
wg.Add(1)
defer rows.Close()
go func(rawSQL string, query *Query, queryResult *QueryResult) {
defer wg.Done()
session := e.engine.NewSession()
defer session.Close()
db := session.DB()
format := query.Model.Get("format").MustString("time_series")
switch format {
case "time_series":
err := e.transformToTimeSeries(query, rows, queryResult, tsdbQuery)
rows, err := db.Query(rawSQL)
if err != nil {
queryResult.Error = err
continue
return
}
case "table":
err := e.transformToTable(query, rows, queryResult, tsdbQuery)
if err != nil {
queryResult.Error = err
continue
defer rows.Close()
format := query.Model.Get("format").MustString("time_series")
switch format {
case "time_series":
err := e.transformToTimeSeries(query, rows, queryResult, tsdbQuery)
if err != nil {
queryResult.Error = err
return
}
case "table":
err := e.transformToTable(query, rows, queryResult, tsdbQuery)
if err != nil {
queryResult.Error = err
return
}
}
}
}(rawSQL, query, queryResult)
}
wg.Wait()
return result, nil
}