diff --git a/conf/defaults.ini b/conf/defaults.ini index a1d44ee4aa6..583a3d90947 100644 --- a/conf/defaults.ini +++ b/conf/defaults.ini @@ -175,6 +175,9 @@ send_user_header = false # Limit the amount of bytes that will be read/accepted from responses of outgoing HTTP requests. response_limit = 0 +# Limits the number of rows that Grafana will process from SQL data sources. +row_limit = 1000000 + #################################### Analytics ########################### [analytics] # Server reporting, sends usage counters to stats.grafana.org every 24 hours. diff --git a/conf/sample.ini b/conf/sample.ini index 7d27f359ffb..1c75d145d2f 100644 --- a/conf/sample.ini +++ b/conf/sample.ini @@ -181,6 +181,9 @@ # Limit the amount of bytes that will be read/accepted from responses of outgoing HTTP requests. ;response_limit = 0 +# Limits the number of rows that Grafana will process from SQL data sources. +;row_limit = 1000000 + #################################### Analytics #################################### [analytics] # Server reporting, sends usage counters to stats.grafana.org every 24 hours. diff --git a/docs/sources/administration/configuration.md b/docs/sources/administration/configuration.md index 902f704ebb5..fc857cfb087 100644 --- a/docs/sources/administration/configuration.md +++ b/docs/sources/administration/configuration.md @@ -439,6 +439,10 @@ If enabled and user is not anonymous, data proxy will add X-Grafana-User header Limits the amount of bytes that will be read/accepted from responses of outgoing HTTP requests. Default is `0` which means disabled. +### row_limit + +Limits the number of rows that Grafana will process from SQL (relational) data sources. Default is `1000000`. +
## [analytics] diff --git a/pkg/setting/setting.go b/pkg/setting/setting.go index 95a2581f4b6..af950175ee8 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -327,6 +327,7 @@ type Cfg struct { DataProxyKeepAlive int DataProxyIdleConnTimeout int ResponseLimit int64 + DataProxyRowLimit int64 // DistributedCache RemoteCacheOptions *RemoteCacheOptions diff --git a/pkg/setting/setting_data_proxy.go b/pkg/setting/setting_data_proxy.go index 463dae09a0b..f879b4c9093 100644 --- a/pkg/setting/setting_data_proxy.go +++ b/pkg/setting/setting_data_proxy.go @@ -2,6 +2,8 @@ package setting import "gopkg.in/ini.v1" +const defaultDataProxyRowLimit = int64(1000000) + func readDataProxySettings(iniFile *ini.File, cfg *Cfg) error { dataproxy := iniFile.Section("dataproxy") cfg.SendUserHeader = dataproxy.Key("send_user_header").MustBool(false) @@ -15,6 +17,11 @@ func readDataProxySettings(iniFile *ini.File, cfg *Cfg) error { cfg.DataProxyMaxIdleConns = dataproxy.Key("max_idle_connections").MustInt() cfg.DataProxyIdleConnTimeout = dataproxy.Key("idle_conn_timeout_seconds").MustInt(90) cfg.ResponseLimit = dataproxy.Key("response_limit").MustInt64(0) + cfg.DataProxyRowLimit = dataproxy.Key("row_limit").MustInt64(defaultDataProxyRowLimit) + + if cfg.DataProxyRowLimit <= 0 { + cfg.DataProxyRowLimit = defaultDataProxyRowLimit + } if val, err := dataproxy.Key("max_idle_connections_per_host").Int(); err == nil { cfg.Logger.Warn("[Deprecated] the configuration setting 'max_idle_connections_per_host' is deprecated, please use 'max_idle_connections' instead") diff --git a/pkg/tsdb/mssql/mssql.go b/pkg/tsdb/mssql/mssql.go index 8ad88f0696a..ead5ff3ed71 100644 --- a/pkg/tsdb/mssql/mssql.go +++ b/pkg/tsdb/mssql/mssql.go @@ -31,9 +31,9 @@ type Service struct { im instancemgmt.InstanceManager } -func ProvideService(manager backendplugin.Manager) (*Service, error) { +func ProvideService(cfg *setting.Cfg, manager backendplugin.Manager) (*Service, error) { s := &Service{ - im: datasource.NewInstanceManager(newInstanceSettings()), + im: datasource.NewInstanceManager(newInstanceSettings(cfg)), } factory := coreplugin.New(backend.ServeOpts{ QueryDataHandler: s, @@ -62,7 +62,7 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest) return dsHandler.QueryData(ctx, req) } -func newInstanceSettings() datasource.InstanceFactoryFunc { +func newInstanceSettings(cfg *setting.Cfg) datasource.InstanceFactoryFunc { return func(settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) { jsonData := sqleng.JsonData{ MaxOpenConns: 0, @@ -89,8 +89,8 @@ func newInstanceSettings() datasource.InstanceFactoryFunc { if err != nil { return nil, err } - // TODO: Don't use global - if setting.Env == setting.Dev { + + if cfg.Env == setting.Dev { logger.Debug("getEngine", "connection", cnnstr) } @@ -99,6 +99,7 @@ func newInstanceSettings() datasource.InstanceFactoryFunc { ConnectionString: cnnstr, DSInfo: dsInfo, MetricColumnTypes: []string{"VARCHAR", "CHAR", "NVARCHAR", "NCHAR"}, + RowLimit: cfg.DataProxyRowLimit, } queryResultTransformer := mssqlQueryResultTransformer{ diff --git a/pkg/tsdb/mssql/mssql_test.go b/pkg/tsdb/mssql/mssql_test.go index d0d4697c941..31ac7602d11 100644 --- a/pkg/tsdb/mssql/mssql_test.go +++ b/pkg/tsdb/mssql/mssql_test.go @@ -57,6 +57,7 @@ func TestMSSQL(t *testing.T) { ConnectionString: "", DSInfo: dsInfo, MetricColumnTypes: []string{"VARCHAR", "CHAR", "NVARCHAR", "NCHAR"}, + RowLimit: 1000000, } endpoint, err := sqleng.NewQueryDataHandler(config, &queryResultTransformer, newMssqlMacroEngine(), logger) require.NoError(t, err) @@ -794,6 +795,7 @@ func TestMSSQL(t *testing.T) { ConnectionString: "", DSInfo: dsInfo, MetricColumnTypes: []string{"VARCHAR", "CHAR", "NVARCHAR", "NCHAR"}, + RowLimit: 1000000, } endpoint, err := sqleng.NewQueryDataHandler(config, &queryResultTransformer, newMssqlMacroEngine(), logger) require.NoError(t, err) @@ -1189,6 +1191,83 @@ func TestMSSQL(t *testing.T) { require.Equal(t, data.FieldTypeNullableTime, frames[0].Fields[0].Type()) require.Equal(t, data.FieldTypeNullableTime, frames[0].Fields[1].Type()) }) + + t.Run("When row limit set to 1", func(t *testing.T) { + queryResultTransformer := mssqlQueryResultTransformer{ + log: logger, + } + dsInfo := sqleng.DataSourceInfo{} + config := sqleng.DataPluginConfiguration{ + DriverName: "mssql", + ConnectionString: "", + DSInfo: dsInfo, + MetricColumnTypes: []string{"VARCHAR", "CHAR", "NVARCHAR", "NCHAR"}, + RowLimit: 1, + } + + handler, err := sqleng.NewQueryDataHandler(config, &queryResultTransformer, newMssqlMacroEngine(), logger) + require.NoError(t, err) + + t.Run("When doing a table query that returns 2 rows should limit the result to 1 row", func(t *testing.T) { + query := &backend.QueryDataRequest{ + Queries: []backend.DataQuery{ + { + JSON: []byte(`{ + "rawSql": "SELECT 1 as value UNION ALL select 2 as value", + "format": "table" + }`), + RefID: "A", + TimeRange: backend.TimeRange{ + From: time.Now(), + To: time.Now(), + }, + }, + }, + } + + resp, err := handler.QueryData(context.Background(), query) + require.NoError(t, err) + queryResult := resp.Responses["A"] + require.NoError(t, queryResult.Error) + frames := queryResult.Frames + require.NoError(t, err) + require.Equal(t, 1, len(frames)) + require.Equal(t, 1, len(frames[0].Fields)) + require.Equal(t, 1, frames[0].Rows()) + require.Len(t, frames[0].Meta.Notices, 1) + require.Equal(t, data.NoticeSeverityWarning, frames[0].Meta.Notices[0].Severity) + }) + + t.Run("When doing a time series that returns 2 rows should limit the result to 1 row", func(t *testing.T) { + query := &backend.QueryDataRequest{ + Queries: []backend.DataQuery{ + { + JSON: []byte(`{ + "rawSql": "SELECT 1 as time, 1 as value UNION ALL select 2 as time, 2 as value", + "format": "time_series" + }`), + RefID: "A", + TimeRange: backend.TimeRange{ + From: time.Now(), + To: time.Now(), + }, + }, + }, + } + + resp, err := handler.QueryData(context.Background(), query) + require.NoError(t, err) + queryResult := resp.Responses["A"] + require.NoError(t, queryResult.Error) + frames := queryResult.Frames + require.NoError(t, err) + require.Equal(t, 1, len(frames)) + require.Equal(t, 2, len(frames[0].Fields)) + require.Equal(t, 1, frames[0].Rows()) + require.Len(t, frames[0].Meta.Notices, 1) + require.Equal(t, data.NoticeSeverityWarning, frames[0].Meta.Notices[0].Severity) + }) + }) }) } diff --git a/pkg/tsdb/mysql/mysql.go b/pkg/tsdb/mysql/mysql.go index 01ddf0b02f9..531a97f6a9f 100644 --- a/pkg/tsdb/mysql/mysql.go +++ b/pkg/tsdb/mysql/mysql.go @@ -46,8 +46,7 @@ func characterEscape(s string, escapeChar string) string { func ProvideService(cfg *setting.Cfg, manager backendplugin.Manager, httpClientProvider httpclient.Provider) (*Service, error) { s := &Service{ - Cfg: cfg, - im: datasource.NewInstanceManager(newInstanceSettings(httpClientProvider)), + im: datasource.NewInstanceManager(newInstanceSettings(cfg, httpClientProvider)), } factory := coreplugin.New(backend.ServeOpts{ QueryDataHandler: s, @@ -59,7 +58,7 @@ func ProvideService(cfg *setting.Cfg, manager backendplugin.Manager, httpClientP return s, nil } -func newInstanceSettings(httpClientProvider httpclient.Provider) datasource.InstanceFactoryFunc { +func newInstanceSettings(cfg *setting.Cfg, httpClientProvider httpclient.Provider) datasource.InstanceFactoryFunc { return func(settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) { jsonData := sqleng.JsonData{ MaxOpenConns: 0, @@ -117,7 +116,7 @@ func newInstanceSettings(httpClientProvider httpclient.Provider) datasource.Inst cnnstr += fmt.Sprintf("&time_zone='%s'", url.QueryEscape(dsInfo.JsonData.Timezone)) } - if setting.Env == setting.Dev { + if cfg.Env == setting.Dev { logger.Debug("getEngine", "connection", cnnstr) } @@ -127,6 +126,7 @@ func newInstanceSettings(httpClientProvider httpclient.Provider) datasource.Inst DSInfo: dsInfo, TimeColumnNames: []string{"time", "time_sec"}, MetricColumnTypes: []string{"CHAR", "VARCHAR", "TINYTEXT", "TEXT", "MEDIUMTEXT", "LONGTEXT"}, + RowLimit: cfg.DataProxyRowLimit, } rowTransformer := mysqlQueryResultTransformer{ diff --git a/pkg/tsdb/mysql/mysql_test.go b/pkg/tsdb/mysql/mysql_test.go index c414329e502..8f021957a4a 100644 --- a/pkg/tsdb/mysql/mysql_test.go +++ b/pkg/tsdb/mysql/mysql_test.go @@ -68,6 +68,7 @@ func TestMySQL(t *testing.T) { DSInfo: dsInfo, TimeColumnNames: []string{"time", "time_sec"}, MetricColumnTypes: []string{"CHAR", "VARCHAR", "TINYTEXT", "TEXT", "MEDIUMTEXT", "LONGTEXT"}, + RowLimit: 1000000, } rowTransformer := mysqlQueryResultTransformer{ @@ -1151,6 +1152,85 @@ func TestMySQL(t *testing.T) { require.Equal(t, data.FieldTypeNullableTime, frames[0].Fields[0].Type()) require.Equal(t, data.FieldTypeNullableTime, frames[0].Fields[1].Type()) }) + + t.Run("When row limit set to 1", func(t *testing.T) { + dsInfo := sqleng.DataSourceInfo{} + config := sqleng.DataPluginConfiguration{ + DriverName: "mysql", + ConnectionString: "", + DSInfo: dsInfo, + TimeColumnNames: []string{"time", "time_sec"}, + MetricColumnTypes: []string{"CHAR", "VARCHAR", "TINYTEXT", "TEXT", "MEDIUMTEXT", "LONGTEXT"}, + RowLimit: 1, + } + + queryResultTransformer := mysqlQueryResultTransformer{ + log: logger, + } + + handler, err := sqleng.NewQueryDataHandler(config, &queryResultTransformer, newMysqlMacroEngine(logger), logger) + require.NoError(t, err) + + t.Run("When doing a table query that returns 2 rows should limit the result to 1 row", func(t *testing.T) { + query := &backend.QueryDataRequest{ + Queries: []backend.DataQuery{ + { + JSON: []byte(`{ + "rawSql": "SELECT 1 as value UNION ALL select 2 as value", + "format": "table" + }`), + RefID: "A", + TimeRange: backend.TimeRange{ + From: time.Now(), + To: time.Now(), + }, + }, + }, + } + + resp, err := handler.QueryData(context.Background(), query) + require.NoError(t, err) + queryResult := resp.Responses["A"] + require.NoError(t, queryResult.Error) + frames := queryResult.Frames + require.NoError(t, err) + require.Equal(t, 1, len(frames)) + require.Equal(t, 1, len(frames[0].Fields)) + require.Equal(t, 1, frames[0].Rows()) + require.Len(t, frames[0].Meta.Notices, 1) + require.Equal(t, data.NoticeSeverityWarning, frames[0].Meta.Notices[0].Severity) + }) + + t.Run("When doing a time series that returns 2 rows should limit the result to 1 row", func(t *testing.T) { + query := &backend.QueryDataRequest{ + Queries: []backend.DataQuery{ + { + JSON: []byte(`{ + "rawSql": "SELECT 1 as time, 1 as value UNION ALL select 2 as time, 2 as value", + "format": "time_series" + }`), + RefID: "A", + TimeRange: backend.TimeRange{ + From: time.Now(), + To: time.Now(), + }, + }, + }, + } + + resp, err := handler.QueryData(context.Background(), query) + require.NoError(t, err) + queryResult := resp.Responses["A"] + require.NoError(t, queryResult.Error) + frames := queryResult.Frames + require.NoError(t, err) + require.Equal(t, 1, len(frames)) + require.Equal(t, 2, len(frames[0].Fields)) + require.Equal(t, 1, frames[0].Rows()) + require.Len(t, frames[0].Meta.Notices, 1) + require.Equal(t, data.NoticeSeverityWarning, frames[0].Meta.Notices[0].Severity) + }) + }) }) } diff --git a/pkg/tsdb/postgres/postgres.go b/pkg/tsdb/postgres/postgres.go index f4ce3bbf9b0..16af7c15ad8 100644 --- a/pkg/tsdb/postgres/postgres.go +++ b/pkg/tsdb/postgres/postgres.go @@ -26,10 +26,9 @@ var logger = log.New("tsdb.postgres") func ProvideService(cfg *setting.Cfg, manager backendplugin.Manager) (*Service, error) { s := &Service{ - Cfg: cfg, tlsManager: newTLSManager(logger, cfg.DataPath), } - s.im = datasource.NewInstanceManager(s.newInstanceSettings()) + s.im = datasource.NewInstanceManager(s.newInstanceSettings(cfg)) factory := coreplugin.New(backend.ServeOpts{ QueryDataHandler: s, }) @@ -41,7 +40,6 @@ func ProvideService(cfg *setting.Cfg, manager backendplugin.Manager) (*Service, } type Service struct { - Cfg *setting.Cfg tlsManager tlsSettingsProvider im instancemgmt.InstanceManager } @@ -63,7 +61,7 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest) return dsInfo.QueryData(ctx, req) } -func (s *Service) newInstanceSettings() datasource.InstanceFactoryFunc { +func (s *Service) newInstanceSettings(cfg *setting.Cfg) datasource.InstanceFactoryFunc { return func(settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) { logger.Debug("Creating Postgres query endpoint") jsonData := sqleng.JsonData{ @@ -94,7 +92,7 @@ func (s *Service) newInstanceSettings() datasource.InstanceFactoryFunc { return nil, err } - if s.Cfg.Env == setting.Dev { + if cfg.Env == setting.Dev { logger.Debug("getEngine", "connection", cnnstr) } @@ -103,6 +101,7 @@ func (s *Service) newInstanceSettings() datasource.InstanceFactoryFunc { ConnectionString: cnnstr, DSInfo: dsInfo, MetricColumnTypes: []string{"UNKNOWN", "TEXT", "VARCHAR", "CHAR"}, + RowLimit: cfg.DataProxyRowLimit, } queryResultTransformer := postgresQueryResultTransformer{ diff --git a/pkg/tsdb/postgres/postgres_test.go b/pkg/tsdb/postgres/postgres_test.go index 46273d66c28..ae9e4baa2a9 100644 --- a/pkg/tsdb/postgres/postgres_test.go +++ b/pkg/tsdb/postgres/postgres_test.go @@ -112,7 +112,6 @@ func TestGenerateConnectionString(t *testing.T) { for _, tt := range testCases { t.Run(tt.desc, func(t *testing.T) { svc := Service{ - Cfg: cfg, tlsManager: &tlsTestManager{settings: tt.tlsSettings}, } @@ -190,6 +189,7 @@ func TestPostgres(t *testing.T) { ConnectionString: "", DSInfo: dsInfo, MetricColumnTypes: []string{"UNKNOWN", "TEXT", "VARCHAR", "CHAR"}, + RowLimit: 1000000, } queryResultTransformer := postgresQueryResultTransformer{ @@ -1225,6 +1225,84 @@ func TestPostgres(t *testing.T) { require.Equal(t, data.FieldTypeNullableTime, frames[0].Fields[0].Type()) require.Equal(t, data.FieldTypeNullableTime, frames[0].Fields[1].Type()) }) + + t.Run("When row limit set to 1", func(t *testing.T) { + dsInfo := sqleng.DataSourceInfo{} + config := sqleng.DataPluginConfiguration{ + DriverName: "postgres", + ConnectionString: "", + DSInfo: dsInfo, + MetricColumnTypes: []string{"UNKNOWN", "TEXT", "VARCHAR", "CHAR"}, + RowLimit: 1, + } + + queryResultTransformer := postgresQueryResultTransformer{ + log: logger, + } + + handler, err := sqleng.NewQueryDataHandler(config, &queryResultTransformer, newPostgresMacroEngine(false), logger) + require.NoError(t, err) + + t.Run("When doing a table query that returns 2 rows should limit the result to 1 row", func(t *testing.T) { + query := &backend.QueryDataRequest{ + Queries: []backend.DataQuery{ + { + JSON: []byte(`{ + "rawSql": "SELECT 1 as value UNION ALL select 2 as value", + "format": "table" + }`), + RefID: "A", + TimeRange: backend.TimeRange{ + From: time.Now(), + To: time.Now(), + }, + }, + }, + } + + resp, err := handler.QueryData(context.Background(), query) + require.NoError(t, err) + queryResult := resp.Responses["A"] + require.NoError(t, queryResult.Error) + frames := queryResult.Frames + require.NoError(t, err) + require.Equal(t, 1, len(frames)) + require.Equal(t, 1, len(frames[0].Fields)) + require.Equal(t, 1, frames[0].Rows()) + require.Len(t, frames[0].Meta.Notices, 1) + require.Equal(t, data.NoticeSeverityWarning, frames[0].Meta.Notices[0].Severity) + }) + + t.Run("When doing a time series query that returns 2 rows should limit the result to 1 row", func(t *testing.T) { + query := &backend.QueryDataRequest{ + Queries: []backend.DataQuery{ + { + JSON: []byte(`{ + "rawSql": "SELECT 1 as time, 1 as value UNION ALL select 2 as time, 2 as value", + "format": "time_series" + }`), + RefID: "A", + TimeRange: backend.TimeRange{ + From: time.Now(), + To: time.Now(), + }, + }, + }, + } + + resp, err := handler.QueryData(context.Background(), query) + require.NoError(t, err) + queryResult := resp.Responses["A"] + require.NoError(t, queryResult.Error) + frames := queryResult.Frames + require.NoError(t, err) + require.Equal(t, 1, len(frames)) + require.Equal(t, 2, len(frames[0].Fields)) + require.Equal(t, 1, frames[0].Rows()) + require.Len(t, frames[0].Meta.Notices, 1) + require.Equal(t, data.NoticeSeverityWarning, frames[0].Meta.Notices[0].Severity) + }) + }) }) } diff --git a/pkg/tsdb/sqleng/sql_engine.go b/pkg/tsdb/sqleng/sql_engine.go index 5bfb5acbc87..f4745aa5cc6 100644 --- a/pkg/tsdb/sqleng/sql_engine.go +++ b/pkg/tsdb/sqleng/sql_engine.go @@ -92,6 +92,7 @@ type DataPluginConfiguration struct { ConnectionString string TimeColumnNames []string MetricColumnTypes []string + RowLimit int64 } type DataSourceHandler struct { macroEngine SQLMacroEngine @@ -101,6 +102,7 @@ type DataSourceHandler struct { metricColumnTypes []string log log.Logger dsInfo DataSourceInfo + rowLimit int64 } type QueryJson struct { RawSql string `json:"rawSql"` @@ -133,6 +135,7 @@ func NewQueryDataHandler(config DataPluginConfiguration, queryResultTransformer timeColumnNames: []string{"time"}, log: log, dsInfo: config.DSInfo, + rowLimit: config.RowLimit, } if len(config.TimeColumnNames) > 0 { @@ -168,8 +171,6 @@ func NewQueryDataHandler(config DataPluginConfiguration, queryResultTransformer return &queryDataHandler, nil } -const rowLimit = 1000000 - type DBDataResponse struct { dataResponse backend.DataResponse refID string @@ -284,15 +285,17 @@ func (e *DataSourceHandler) executeQuery(query backend.DataQuery, wg *sync.WaitG // Convert row.Rows to dataframe stringConverters := e.queryResultTransformer.GetConverterList() - frame, err := sqlutil.FrameFromRows(rows.Rows, rowLimit, sqlutil.ToConverters(stringConverters...)...) + frame, err := sqlutil.FrameFromRows(rows.Rows, e.rowLimit, sqlutil.ToConverters(stringConverters...)...) if err != nil { errAppendDebug("convert frame from rows error", err, interpolatedQuery) return } - frame.SetMeta(&data.FrameMeta{ - ExecutedQueryString: interpolatedQuery, - }) + if frame.Meta == nil { + frame.Meta = &data.FrameMeta{} + } + + frame.Meta.ExecutedQueryString = interpolatedQuery // If no rows were returned, no point checking anything else. if frame.Rows() == 0 {