Postgres Data Source (#9475)

* add postgresql datasource * add rest of files for postgres datasource * fix timeseries query, remove unused code * consistent naming, refactoring * s/mysql/postgres/ * s/mysql/postgres/ * couple more tests * tests for more datatypes * fix macros for postgres * add __timeSec macro * add frontend for postgres datasource * adjust documentation * fix formatting * add proper plugin description * merge editor changes from mysql * port changes from mysql datasource * set proper defaultQuery for postgres * add time_sec to timeseries query accept int for value for timeseries query * revert allowing time_sec and handle int or float values as unix timestamp for "time" column * fix tslint error * handle decimal values in timeseries query * allow setting sslmode for postgres datasource * use type switch for handling data types * fix value for timeseries query * refactor timeseries queries to make them more flexible * remove debug statement from inner loop in type conversion * use plain for loop in getTypedRowData * fix timeseries queries * adjust postgres datasource to tsdb refactoring * adjust postgres datasource to frontend changes * update lib/pq to latest version * move type conversion to getTypedRowData * handle address types cidr, inet and macaddr * adjust response parser and docs for annotations * convert unknown types to string * add documentation for postgres datasource * add another example query with metric column * set more helpful default query * update help text in query editor * handle NULL in value column of timeseries query * add __timeGroup macro * add test for __timeGroup macro * document __timeGroup and set proper default query for annotations * fix typos in docs * add postgres to list of datasources * add postgres to builtInPlugins * mysql: refactoring as prep for merging postgres Refactors out the initialization of the xorm engine and the query logic for an sql data source. * mysql: rename refactoring + test update * postgres:refactor to use SqlEngine(same as mysql) Refactored to use a common base class with the MySql data source. Other changes from the original PR: - Changed time column to be time_sec to allow other time units in the future and to be the same as MySQL - Changed integration test to test the main Query method rather than the private transformToTable method - Changed the __timeSec macro name to __timeEpoch - Renamed PostgresExecutor to PostgresQueryEndpoint Fixes #9209 (the original PR) * postgres: encrypt password on config page With some other cosmetic changes to the config page: - placeholder texts - reset button for the password after it has been encrypted. - default value for the sslmode field. * postgres: change back col name to time from time_sec * postgres mysql: remove annotation title Title has been removed from annotations * postgres: fix images for docs page * postgres mysql: fix specs
2017-10-10 15:19:14 +02:00
parent 630e6f5da6
commit d1c9760fa8
36 changed files with 2231 additions and 478 deletions
@@ -0,0 +1,99 @@
+package postgres
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/grafana/grafana/pkg/tsdb"
+)
+
+//const rsString = `(?:"([^"]*)")`;
+const rsIdentifier = `([_a-zA-Z0-9]+)`
+const sExpr = `\$` + rsIdentifier + `\(([^\)]*)\)`
+
+type PostgresMacroEngine struct {
+	TimeRange *tsdb.TimeRange
+}
+
+func NewPostgresMacroEngine() tsdb.SqlMacroEngine {
+	return &PostgresMacroEngine{}
+}
+
+func (m *PostgresMacroEngine) Interpolate(timeRange *tsdb.TimeRange, sql string) (string, error) {
+	m.TimeRange = timeRange
+	rExp, _ := regexp.Compile(sExpr)
+	var macroError error
+
+	sql = replaceAllStringSubmatchFunc(rExp, sql, func(groups []string) string {
+		res, err := m.evaluateMacro(groups[1], strings.Split(groups[2], ","))
+		if err != nil && macroError == nil {
+			macroError = err
+			return "macro_error()"
+		}
+		return res
+	})
+
+	if macroError != nil {
+		return "", macroError
+	}
+
+	return sql, nil
+}
+
+func replaceAllStringSubmatchFunc(re *regexp.Regexp, str string, repl func([]string) string) string {
+	result := ""
+	lastIndex := 0
+
+	for _, v := range re.FindAllSubmatchIndex([]byte(str), -1) {
+		groups := []string{}
+		for i := 0; i < len(v); i += 2 {
+			groups = append(groups, str[v[i]:v[i+1]])
+		}
+
+		result += str[lastIndex:v[0]] + repl(groups)
+		lastIndex = v[1]
+	}
+
+	return result + str[lastIndex:]
+}
+
+func (m *PostgresMacroEngine) evaluateMacro(name string, args []string) (string, error) {
+	switch name {
+	case "__time":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+		return fmt.Sprintf("%s AS \"time\"", args[0]), nil
+	case "__timeEpoch":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+		return fmt.Sprintf("extract(epoch from %s) as \"time\"", args[0]), nil
+	case "__timeFilter":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+		return fmt.Sprintf("%s >= to_timestamp(%d) AND %s <= to_timestamp(%d)", args[0], uint64(m.TimeRange.GetFromAsMsEpoch()/1000), args[0], uint64(m.TimeRange.GetToAsMsEpoch()/1000)), nil
+	case "__timeFrom":
+		return fmt.Sprintf("to_timestamp(%d)", uint64(m.TimeRange.GetFromAsMsEpoch()/1000)), nil
+	case "__timeTo":
+		return fmt.Sprintf("to_timestamp(%d)", uint64(m.TimeRange.GetToAsMsEpoch()/1000)), nil
+	case "__timeGroup":
+		if len(args) < 2 {
+			return "", fmt.Errorf("macro %v needs time column and interval", name)
+		}
+		return fmt.Sprintf("(extract(epoch from \"%s\")/extract(epoch from %s::interval))::int", args[0], args[1]), nil
+	case "__unixEpochFilter":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+		return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], uint64(m.TimeRange.GetFromAsMsEpoch()/1000), args[0], uint64(m.TimeRange.GetToAsMsEpoch()/1000)), nil
+	case "__unixEpochFrom":
+		return fmt.Sprintf("%d", uint64(m.TimeRange.GetFromAsMsEpoch()/1000)), nil
+	case "__unixEpochTo":
+		return fmt.Sprintf("%d", uint64(m.TimeRange.GetToAsMsEpoch()/1000)), nil
+	default:
+		return "", fmt.Errorf("Unknown macro %v", name)
+	}
+}
@@ -0,0 +1,80 @@
+package postgres
+
+import (
+	"testing"
+
+	"github.com/grafana/grafana/pkg/tsdb"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestMacroEngine(t *testing.T) {
+	Convey("MacroEngine", t, func() {
+		engine := &PostgresMacroEngine{}
+		timeRange := &tsdb.TimeRange{From: "5m", To: "now"}
+
+		Convey("interpolate __time function", func() {
+			sql, err := engine.Interpolate(nil, "select $__time(time_column)")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "select time_column AS \"time\"")
+		})
+
+		Convey("interpolate __time function wrapped in aggregation", func() {
+			sql, err := engine.Interpolate(nil, "select min($__time(time_column))")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "select min(time_column AS \"time\")")
+		})
+
+		Convey("interpolate __timeFilter function", func() {
+			sql, err := engine.Interpolate(timeRange, "WHERE $__timeFilter(time_column)")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "WHERE time_column >= to_timestamp(18446744066914186738) AND time_column <= to_timestamp(18446744066914187038)")
+		})
+
+		Convey("interpolate __timeFrom function", func() {
+			sql, err := engine.Interpolate(timeRange, "select $__timeFrom(time_column)")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "select to_timestamp(18446744066914186738)")
+		})
+
+		Convey("interpolate __timeGroup function", func() {
+
+			sql, err := engine.Interpolate(timeRange, "GROUP BY $__timeGroup(time_column,'5m')")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "GROUP BY (extract(epoch from \"time_column\")/extract(epoch from '5m'::interval))::int")
+		})
+
+		Convey("interpolate __timeTo function", func() {
+			sql, err := engine.Interpolate(timeRange, "select $__timeTo(time_column)")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "select to_timestamp(18446744066914187038)")
+		})
+
+		Convey("interpolate __unixEpochFilter function", func() {
+			sql, err := engine.Interpolate(timeRange, "select $__unixEpochFilter(18446744066914186738)")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "select 18446744066914186738 >= 18446744066914186738 AND 18446744066914186738 <= 18446744066914187038")
+		})
+
+		Convey("interpolate __unixEpochFrom function", func() {
+			sql, err := engine.Interpolate(timeRange, "select $__unixEpochFrom()")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "select 18446744066914186738")
+		})
+
+		Convey("interpolate __unixEpochTo function", func() {
+			sql, err := engine.Interpolate(timeRange, "select $__unixEpochTo()")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "select 18446744066914187038")
+		})
+
+	})
+}
@@ -0,0 +1,245 @@
+package postgres
+
+import (
+	"container/list"
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/go-xorm/core"
+	"github.com/grafana/grafana/pkg/components/null"
+	"github.com/grafana/grafana/pkg/log"
+	"github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/tsdb"
+)
+
+type PostgresQueryEndpoint struct {
+	sqlEngine tsdb.SqlEngine
+	log       log.Logger
+}
+
+func init() {
+	tsdb.RegisterTsdbQueryEndpoint("postgres", NewPostgresQueryEndpoint)
+}
+
+func NewPostgresQueryEndpoint(datasource *models.DataSource) (tsdb.TsdbQueryEndpoint, error) {
+	endpoint := &PostgresQueryEndpoint{
+		log: log.New("tsdb.postgres"),
+	}
+
+	endpoint.sqlEngine = &tsdb.DefaultSqlEngine{
+		MacroEngine: NewPostgresMacroEngine(),
+	}
+
+	cnnstr := generateConnectionString(datasource)
+	endpoint.log.Debug("getEngine", "connection", cnnstr)
+
+	if err := endpoint.sqlEngine.InitEngine("postgres", datasource, cnnstr); err != nil {
+		return nil, err
+	}
+
+	return endpoint, nil
+}
+
+func generateConnectionString(datasource *models.DataSource) string {
+	password := ""
+	for key, value := range datasource.SecureJsonData.Decrypt() {
+		if key == "password" {
+			password = value
+			break
+		}
+	}
+
+	sslmode := datasource.JsonData.Get("sslmode").MustString("require")
+	return fmt.Sprintf("postgres://%s:%s@%s/%s?sslmode=%s", datasource.User, password, datasource.Url, datasource.Database, sslmode)
+}
+
+func (e *PostgresQueryEndpoint) Query(ctx context.Context, dsInfo *models.DataSource, tsdbQuery *tsdb.TsdbQuery) (*tsdb.Response, error) {
+	return e.sqlEngine.Query(ctx, dsInfo, tsdbQuery, e.transformToTimeSeries, e.transformToTable)
+}
+
+func (e PostgresQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
+
+	columnNames, err := rows.Columns()
+	if err != nil {
+		return err
+	}
+
+	table := &tsdb.Table{
+		Columns: make([]tsdb.TableColumn, len(columnNames)),
+		Rows:    make([]tsdb.RowValues, 0),
+	}
+
+	for i, name := range columnNames {
+		table.Columns[i].Text = name
+	}
+
+	rowLimit := 1000000
+	rowCount := 0
+
+	for ; rows.Next(); rowCount++ {
+		if rowCount > rowLimit {
+			return fmt.Errorf("PostgreSQL query row limit exceeded, limit %d", rowLimit)
+		}
+
+		values, err := e.getTypedRowData(rows)
+		if err != nil {
+			return err
+		}
+
+		table.Rows = append(table.Rows, values)
+	}
+
+	result.Tables = append(result.Tables, table)
+	result.Meta.Set("rowCount", rowCount)
+	return nil
+}
+
+func (e PostgresQueryEndpoint) getTypedRowData(rows *core.Rows) (tsdb.RowValues, error) {
+
+	types, err := rows.ColumnTypes()
+	if err != nil {
+		return nil, err
+	}
+
+	values := make([]interface{}, len(types))
+	valuePtrs := make([]interface{}, len(types))
+
+	for i := 0; i < len(types); i++ {
+		valuePtrs[i] = &values[i]
+	}
+
+	if err := rows.Scan(valuePtrs...); err != nil {
+		return nil, err
+	}
+
+	// convert types not handled by lib/pq
+	// unhandled types are returned as []byte
+	for i := 0; i < len(types); i++ {
+		if value, ok := values[i].([]byte); ok == true {
+			switch types[i].DatabaseTypeName() {
+			case "NUMERIC":
+				if v, err := strconv.ParseFloat(string(value), 64); err == nil {
+					values[i] = v
+				} else {
+					e.log.Debug("Rows", "Error converting numeric to float", value)
+				}
+			case "UNKNOWN", "CIDR", "INET", "MACADDR":
+				// char literals have type UNKNOWN
+				values[i] = string(value)
+			default:
+				e.log.Debug("Rows", "Unknown database type", types[i].DatabaseTypeName(), "value", value)
+				values[i] = string(value)
+			}
+		}
+	}
+
+	return values, nil
+}
+
+func (e PostgresQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
+	pointsBySeries := make(map[string]*tsdb.TimeSeries)
+	seriesByQueryOrder := list.New()
+	columnNames, err := rows.Columns()
+
+	if err != nil {
+		return err
+	}
+
+	rowLimit := 1000000
+	rowCount := 0
+	timeIndex := -1
+	metricIndex := -1
+
+	// check columns of resultset
+	for i, col := range columnNames {
+		switch col {
+		case "time":
+			timeIndex = i
+		case "metric":
+			metricIndex = i
+		}
+	}
+
+	if timeIndex == -1 {
+		return fmt.Errorf("Found no column named time")
+	}
+
+	for rows.Next() {
+		var timestamp float64
+		var value null.Float
+		var metric string
+
+		if rowCount > rowLimit {
+			return fmt.Errorf("PostgreSQL query row limit exceeded, limit %d", rowLimit)
+		}
+
+		values, err := e.getTypedRowData(rows)
+		if err != nil {
+			return err
+		}
+
+		switch columnValue := values[timeIndex].(type) {
+		case int64:
+			timestamp = float64(columnValue * 1000)
+		case float64:
+			timestamp = columnValue * 1000
+		case time.Time:
+			timestamp = float64(columnValue.Unix() * 1000)
+		default:
+			return fmt.Errorf("Invalid type for column time, must be of type timestamp or unix timestamp")
+		}
+
+		if metricIndex >= 0 {
+			if columnValue, ok := values[metricIndex].(string); ok == true {
+				metric = columnValue
+			} else {
+				return fmt.Errorf("Column metric must be of type char,varchar or text")
+			}
+		}
+
+		for i, col := range columnNames {
+			if i == timeIndex || i == metricIndex {
+				continue
+			}
+
+			switch columnValue := values[i].(type) {
+			case int64:
+				value = null.FloatFrom(float64(columnValue))
+			case float64:
+				value = null.FloatFrom(columnValue)
+			case nil:
+				value.Valid = false
+			default:
+				return fmt.Errorf("Value column must have numeric datatype, column: %s type: %T value: %v", col, columnValue, columnValue)
+			}
+			if metricIndex == -1 {
+				metric = col
+			}
+			e.appendTimePoint(pointsBySeries, seriesByQueryOrder, metric, timestamp, value)
+			rowCount++
+
+		}
+	}
+
+	for elem := seriesByQueryOrder.Front(); elem != nil; elem = elem.Next() {
+		key := elem.Value.(string)
+		result.Series = append(result.Series, pointsBySeries[key])
+	}
+
+	result.Meta.Set("rowCount", rowCount)
+	return nil
+}
+
+func (e PostgresQueryEndpoint) appendTimePoint(pointsBySeries map[string]*tsdb.TimeSeries, seriesByQueryOrder *list.List, metric string, timestamp float64, value null.Float) {
+	if series, exist := pointsBySeries[metric]; exist {
+		series.Points = append(series.Points, tsdb.TimePoint{value, null.FloatFrom(timestamp)})
+	} else {
+		series := &tsdb.TimeSeries{Name: metric}
+		series.Points = append(series.Points, tsdb.TimePoint{value, null.FloatFrom(timestamp)})
+		pointsBySeries[metric] = series
+		seriesByQueryOrder.PushBack(metric)
+	}
+	e.log.Debug("Rows", "metric", metric, "time", timestamp, "value", value)
+}
@@ -0,0 +1,125 @@
+package postgres
+
+import (
+	"testing"
+	"time"
+
+	"github.com/go-xorm/xorm"
+	"github.com/grafana/grafana/pkg/components/simplejson"
+	"github.com/grafana/grafana/pkg/log"
+	"github.com/grafana/grafana/pkg/services/sqlstore/sqlutil"
+	"github.com/grafana/grafana/pkg/tsdb"
+	_ "github.com/lib/pq"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+// To run this test, remove the Skip from SkipConvey
+// and set up a PostgreSQL db named grafanatest and a user/password grafanatest/grafanatest
+func TestPostgres(t *testing.T) {
+	SkipConvey("PostgreSQL", t, func() {
+		x := InitPostgresTestDB(t)
+
+		endpoint := &PostgresQueryEndpoint{
+			sqlEngine: &tsdb.DefaultSqlEngine{
+				MacroEngine: NewPostgresMacroEngine(),
+				XormEngine:  x,
+			},
+			log: log.New("tsdb.postgres"),
+		}
+
+		sess := x.NewSession()
+		defer sess.Close()
+
+		sql := `
+      CREATE TABLE postgres_types(
+        c00_smallint smallint,
+        c01_integer integer,
+        c02_bigint bigint,
+
+        c03_real real,
+        c04_double double precision,
+        c05_decimal decimal(10,2),
+        c06_numeric numeric(10,2),
+
+        c07_char char(10),
+        c08_varchar varchar(10),
+        c09_text text,
+
+        c10_timestamp timestamp without time zone,
+        c11_timestamptz timestamp with time zone,
+        c12_date date,
+        c13_time time without time zone,
+        c14_timetz time with time zone,
+        c15_interval interval
+      );
+    `
+		_, err := sess.Exec(sql)
+		So(err, ShouldBeNil)
+
+		sql = `
+      INSERT INTO postgres_types VALUES(
+        1,2,3,
+        4.5,6.7,1.1,1.2,
+        'char10','varchar10','text',
+
+        now(),now(),now(),now(),now(),'15m'::interval
+      );
+    `
+		_, err = sess.Exec(sql)
+		So(err, ShouldBeNil)
+
+		Convey("Query with Table format should map PostgreSQL column types to Go types", func() {
+			query := &tsdb.TsdbQuery{
+				Queries: []*tsdb.Query{
+					{
+						Model: simplejson.NewFromAny(map[string]interface{}{
+							"rawSql": "SELECT * FROM postgres_types",
+							"format": "table",
+						}),
+						RefId: "A",
+					},
+				},
+			}
+
+			resp, err := endpoint.Query(nil, nil, query)
+			queryResult := resp.Results["A"]
+			So(err, ShouldBeNil)
+
+			column := queryResult.Tables[0].Rows[0]
+			So(column[0].(int64), ShouldEqual, 1)
+			So(column[1].(int64), ShouldEqual, 2)
+			So(column[2].(int64), ShouldEqual, 3)
+			So(column[3].(float64), ShouldEqual, 4.5)
+			So(column[4].(float64), ShouldEqual, 6.7)
+			// libpq doesnt properly convert decimal, numeric and char to go types but returns []uint8 instead
+			//			So(column[5].(float64), ShouldEqual, 1.1)
+			//			So(column[6].(float64), ShouldEqual, 1.2)
+			//			So(column[7].(string), ShouldEqual, "char")
+			So(column[8].(string), ShouldEqual, "varchar10")
+			So(column[9].(string), ShouldEqual, "text")
+
+			So(column[10].(time.Time), ShouldHaveSameTypeAs, time.Now())
+			So(column[11].(time.Time), ShouldHaveSameTypeAs, time.Now())
+			So(column[12].(time.Time), ShouldHaveSameTypeAs, time.Now())
+			So(column[13].(time.Time), ShouldHaveSameTypeAs, time.Now())
+			So(column[14].(time.Time), ShouldHaveSameTypeAs, time.Now())
+
+			// libpq doesnt properly convert interval to go types but returns []uint8 instead
+			//			So(column[15].(time.Time), ShouldHaveSameTypeAs, time.Now())
+		})
+	})
+}
+
+func InitPostgresTestDB(t *testing.T) *xorm.Engine {
+	x, err := xorm.NewEngine(sqlutil.TestDB_Postgres.DriverName, sqlutil.TestDB_Postgres.ConnStr)
+
+	// x.ShowSQL()
+
+	if err != nil {
+		t.Fatalf("Failed to init postgres db %v", err)
+	}
+
+	sqlutil.CleanDB(x)
+
+	return x
+}