mysql: began work on backend macro engine

pkg/services/sqlstore/migrations/migrations.go

@@ -24,7 +24,6 @@ func AddMigrations(mg *Migrator) {
 	addPreferencesMigrations(mg)
 	addAlertMigrations(mg)
 	addAnnotationMig(mg)
-	addStatsMigrations(mg)
 	addTestDataMigrations(mg)
 }
 

pkg/services/sqlstore/sql_test_data.go

@@ -14,7 +14,7 @@ func init() {
 
 func sqlRandomWalk(m1 string, m2 string, intWalker int64, floatWalker float64, sess *session) error {
 
-	timeWalker := time.Now().UTC().Add(time.Hour * -1)
+	timeWalker := time.Now().UTC().Add(time.Hour * -200)
 	now := time.Now().UTC()
 	step := time.Minute
 
@@ -29,7 +29,7 @@ func sqlRandomWalk(m1 string, m2 string, intWalker int64, floatWalker float64, s
 		timeWalker = timeWalker.Add(step)
 
 		row.Id = 0
-		row.ValueBigInt += rand.Int63n(100) - 100
+		row.ValueBigInt += rand.Int63n(200) - 100
 		row.ValueDouble += rand.Float64() - 0.5
 		row.ValueFloat += rand.Float32() - 0.5
 		row.TimeEpoch = timeWalker.Unix()

pkg/tsdb/mysql/macros.go

@@ -0,0 +1,75 @@
+package mysql
+
+import (
+	"fmt"
+	"regexp"
+
+	"github.com/grafana/grafana/pkg/tsdb"
+)
+
+//const rsString = `(?:"([^"]*)")`;
+const rsIdentifier = `([_a-zA-Z0-9]+)`
+const sExpr = `\$` + rsIdentifier + `\((.*)\)`
+
+type SqlMacroEngine interface {
+	Interpolate(sql string) (string, error)
+}
+
+type MySqlMacroEngine struct {
+	TimeRange *tsdb.TimeRange
+}
+
+func NewMysqlMacroEngine(timeRange *tsdb.TimeRange) SqlMacroEngine {
+	return &MySqlMacroEngine{
+		TimeRange: timeRange,
+	}
+}
+
+func (m *MySqlMacroEngine) Interpolate(sql string) (string, error) {
+	rExp, _ := regexp.Compile(sExpr)
+	var macroError error
+
+	sql = ReplaceAllStringSubmatchFunc(rExp, sql, func(groups []string) string {
+		res, err := m.EvaluateMacro(groups[1], groups[2:len(groups)])
+		if 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 *MySqlMacroEngine) 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 "UNIX_TIMESTAMP(" + args[0] + ") as time_sec", nil
+	default:
+		return "", fmt.Errorf("Unknown macro %v", name)
+	}
+}

pkg/tsdb/mysql/macros_test.go

@@ -0,0 +1,22 @@
+package mysql
+
+import (
+	"testing"
+
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestMacroEngine(t *testing.T) {
+	Convey("MacroEngine", t, func() {
+
+		Convey("interpolate simple function", func() {
+			engine := &MySqlMacroEngine{}
+
+			sql, err := engine.Interpolate("select $__time(time_column)")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "select UNIX_TIMESTAMP(time_column) as time_sec")
+		})
+
+	})
+}

pkg/tsdb/mysql/mysql.go

@@ -99,40 +99,49 @@ func (e *MysqlExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice, co
 
 		defer rows.Close()
 
-		result.QueryResults[query.RefId] = e.TransformToTimeSeries(query, rows)
+		res, err := e.TransformToTimeSeries(query, rows)
+		if err != nil {
+			result.Error = err
+			return result
+		}
+
+		result.QueryResults[query.RefId] = &tsdb.QueryResult{RefId: query.RefId, Series: res}
 	}
 
 	return result
 }
 
-func (e MysqlExecutor) TransformToTimeSeries(query *tsdb.Query, rows *core.Rows) *tsdb.QueryResult {
+func (e MysqlExecutor) TransformToTimeSeries(query *tsdb.Query, rows *core.Rows) (tsdb.TimeSeriesSlice, error) {
-	result := &tsdb.QueryResult{RefId: query.RefId}
 	pointsBySeries := make(map[string]*tsdb.TimeSeries)
 	columnNames, err := rows.Columns()
 
 	if err != nil {
-		result.Error = err
+		return nil, err
-		return result
 	}
 
 	rowData := NewStringStringScan(columnNames)
-	for rows.Next() {
+	rowLimit := 1000000
+	rowCount := 0
+
+	for ; rows.Next(); rowCount += 1 {
+		if rowCount > rowLimit {
+			return nil, fmt.Errorf("MySQL query row limit exceeded, limit %d", rowLimit)
+		}
+
 		err := rowData.Update(rows.Rows)
 		if err != nil {
-			e.log.Error("Mysql response parsing", "error", err)
+			e.log.Error("MySQL response parsing", "error", err)
-			result.Error = err
+			return nil, fmt.Errorf("MySQL response parsing error %v", err)
-			return result
 		}
 
 		if rowData.metric == "" {
 			rowData.metric = "Unknown"
 		}
 
-		e.log.Info("Rows", "metric", rowData.metric, "time", rowData.time, "value", rowData.value)
+		//e.log.Debug("Rows", "metric", rowData.metric, "time", rowData.time, "value", rowData.value)
 
 		if !rowData.time.Valid {
-			result.Error = fmt.Errorf("Found row with no time value")
+			return nil, fmt.Errorf("Found row with no time value")
-			return result
 		}
 
 		if series, exist := pointsBySeries[rowData.metric]; exist {
@@ -144,11 +153,13 @@ func (e MysqlExecutor) TransformToTimeSeries(query *tsdb.Query, rows *core.Rows)
 		}
 	}
 
+	seriesList := make(tsdb.TimeSeriesSlice, 0)
 	for _, value := range pointsBySeries {
-		result.Series = append(result.Series, value)
+		seriesList = append(seriesList, value)
 	}
 
-	return result
+	e.log.Debug("TransformToTimeSeries", "rowCount", rowCount, "timeSeriesCount", len(seriesList))
+	return seriesList, nil
 }
 
 type stringStringScan struct {