diff --git a/pkg/tsdb/mssql/macros.go b/pkg/tsdb/mssql/macros.go index 33cb6e79ba4..812cf45bf5d 100644 --- a/pkg/tsdb/mssql/macros.go +++ b/pkg/tsdb/mssql/macros.go @@ -16,8 +16,6 @@ const sExpr = `\$` + rsIdentifier + `\(([^\)]*)\)` type msSQLMacroEngine struct { *sqleng.SQLMacroEngineBase - timeRange plugins.DataTimeRange - query plugins.DataSubQuery } func newMssqlMacroEngine() sqleng.SQLMacroEngine { @@ -26,8 +24,6 @@ func newMssqlMacroEngine() sqleng.SQLMacroEngine { func (m *msSQLMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange plugins.DataTimeRange, sql string) (string, error) { - m.timeRange = timeRange - m.query = query // TODO: Return any error rExp, _ := regexp.Compile(sExpr) var macroError error @@ -37,7 +33,7 @@ func (m *msSQLMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange plu for i, arg := range args { args[i] = strings.Trim(arg, " ") } - res, err := m.evaluateMacro(groups[1], args) + res, err := m.evaluateMacro(timeRange, query, groups[1], args) if err != nil && macroError == nil { macroError = err return "macro_error()" @@ -52,7 +48,7 @@ func (m *msSQLMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange plu return sql, nil } -func (m *msSQLMacroEngine) evaluateMacro(name string, args []string) (string, error) { +func (m *msSQLMacroEngine) evaluateMacro(timeRange plugins.DataTimeRange, query plugins.DataSubQuery, name string, args []string) (string, error) { switch name { case "__time": if len(args) == 0 { @@ -69,11 +65,11 @@ func (m *msSQLMacroEngine) evaluateMacro(name string, args []string) (string, er return "", fmt.Errorf("missing time column argument for macro %v", name) } - return fmt.Sprintf("%s BETWEEN '%s' AND '%s'", args[0], m.timeRange.GetFromAsTimeUTC().Format(time.RFC3339), m.timeRange.GetToAsTimeUTC().Format(time.RFC3339)), nil + return fmt.Sprintf("%s BETWEEN '%s' AND '%s'", args[0], timeRange.GetFromAsTimeUTC().Format(time.RFC3339), timeRange.GetToAsTimeUTC().Format(time.RFC3339)), nil case "__timeFrom": - return fmt.Sprintf("'%s'", m.timeRange.GetFromAsTimeUTC().Format(time.RFC3339)), nil + return fmt.Sprintf("'%s'", timeRange.GetFromAsTimeUTC().Format(time.RFC3339)), nil case "__timeTo": - return fmt.Sprintf("'%s'", m.timeRange.GetToAsTimeUTC().Format(time.RFC3339)), nil + return fmt.Sprintf("'%s'", timeRange.GetToAsTimeUTC().Format(time.RFC3339)), nil case "__timeGroup": if len(args) < 2 { return "", fmt.Errorf("macro %v needs time column and interval", name) @@ -83,14 +79,14 @@ func (m *msSQLMacroEngine) evaluateMacro(name string, args []string) (string, er return "", fmt.Errorf("error parsing interval %v", args[1]) } if len(args) == 3 { - err := sqleng.SetupFillmode(m.query, interval, args[2]) + err := sqleng.SetupFillmode(query, interval, args[2]) if err != nil { return "", err } } return fmt.Sprintf("FLOOR(DATEDIFF(second, '1970-01-01', %s)/%.0f)*%.0f", args[0], interval.Seconds(), interval.Seconds()), nil case "__timeGroupAlias": - tg, err := m.evaluateMacro("__timeGroup", args) + tg, err := m.evaluateMacro(timeRange, query, "__timeGroup", args) if err == nil { return tg + " AS [time]", nil } @@ -99,16 +95,16 @@ func (m *msSQLMacroEngine) evaluateMacro(name string, args []string) (string, er 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], m.timeRange.GetFromAsSecondsEpoch(), args[0], m.timeRange.GetToAsSecondsEpoch()), nil + return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], timeRange.GetFromAsSecondsEpoch(), args[0], timeRange.GetToAsSecondsEpoch()), nil case "__unixEpochNanoFilter": 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], m.timeRange.GetFromAsTimeUTC().UnixNano(), args[0], m.timeRange.GetToAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], timeRange.GetFromAsTimeUTC().UnixNano(), args[0], timeRange.GetToAsTimeUTC().UnixNano()), nil case "__unixEpochNanoFrom": - return fmt.Sprintf("%d", m.timeRange.GetFromAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%d", timeRange.GetFromAsTimeUTC().UnixNano()), nil case "__unixEpochNanoTo": - return fmt.Sprintf("%d", m.timeRange.GetToAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%d", timeRange.GetToAsTimeUTC().UnixNano()), nil case "__unixEpochGroup": if len(args) < 2 { return "", fmt.Errorf("macro %v needs time column and interval and optional fill value", name) @@ -118,14 +114,14 @@ func (m *msSQLMacroEngine) evaluateMacro(name string, args []string) (string, er return "", fmt.Errorf("error parsing interval %v", args[1]) } if len(args) == 3 { - err := sqleng.SetupFillmode(m.query, interval, args[2]) + err := sqleng.SetupFillmode(query, interval, args[2]) if err != nil { return "", err } } return fmt.Sprintf("FLOOR(%s/%v)*%v", args[0], interval.Seconds(), interval.Seconds()), nil case "__unixEpochGroupAlias": - tg, err := m.evaluateMacro("__unixEpochGroup", args) + tg, err := m.evaluateMacro(timeRange, query, "__unixEpochGroup", args) if err == nil { return tg + " AS [time]", nil } diff --git a/pkg/tsdb/mssql/macros_test.go b/pkg/tsdb/mssql/macros_test.go index 7734d7e0555..9ab9a68e045 100644 --- a/pkg/tsdb/mssql/macros_test.go +++ b/pkg/tsdb/mssql/macros_test.go @@ -3,6 +3,7 @@ package mssql import ( "fmt" "strconv" + "sync" "testing" "time" @@ -10,6 +11,7 @@ import ( "github.com/grafana/grafana/pkg/components/simplejson" "github.com/grafana/grafana/pkg/plugins" . "github.com/smartystreets/goconvey/convey" + "github.com/stretchr/testify/require" ) func TestMacroEngine(t *testing.T) { @@ -224,3 +226,33 @@ func TestMacroEngine(t *testing.T) { }) }) } + +func TestMacroEngineConcurrency(t *testing.T) { + engine := newMssqlMacroEngine() + query1 := plugins.DataSubQuery{ + Model: simplejson.New(), + } + query2 := plugins.DataSubQuery{ + Model: simplejson.New(), + } + from := time.Date(2018, 4, 12, 18, 0, 0, 0, time.UTC) + to := from.Add(5 * time.Minute) + timeRange := plugins.DataTimeRange{From: "5m", To: "now", Now: to} + + var wg sync.WaitGroup + wg.Add(2) + + go func(query plugins.DataSubQuery) { + defer wg.Done() + _, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m')") + require.NoError(t, err) + }(query1) + + go func(query plugins.DataSubQuery) { + _, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m')") + require.NoError(t, err) + defer wg.Done() + }(query2) + + wg.Wait() +} diff --git a/pkg/tsdb/mysql/macros.go b/pkg/tsdb/mysql/macros.go index 9a21885d588..0d70815f2bd 100644 --- a/pkg/tsdb/mysql/macros.go +++ b/pkg/tsdb/mysql/macros.go @@ -19,9 +19,7 @@ var restrictedRegExp = regexp.MustCompile(`(?im)([\s]*show[\s]+grants|[\s,]sessi type mySQLMacroEngine struct { *sqleng.SQLMacroEngineBase - timeRange plugins.DataTimeRange - query plugins.DataSubQuery - logger log.Logger + logger log.Logger } func newMysqlMacroEngine(logger log.Logger) sqleng.SQLMacroEngine { @@ -29,9 +27,6 @@ func newMysqlMacroEngine(logger log.Logger) sqleng.SQLMacroEngine { } func (m *mySQLMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange plugins.DataTimeRange, sql string) (string, error) { - m.timeRange = timeRange - m.query = query - matches := restrictedRegExp.FindAllStringSubmatch(sql, 1) if len(matches) > 0 { m.logger.Error("show grants, session_user(), current_user(), system_user() or user() not allowed in query") @@ -47,7 +42,7 @@ func (m *mySQLMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange plu for i, arg := range args { args[i] = strings.Trim(arg, " ") } - res, err := m.evaluateMacro(groups[1], args) + res, err := m.evaluateMacro(timeRange, query, groups[1], args) if err != nil && macroError == nil { macroError = err return "macro_error()" @@ -62,7 +57,7 @@ func (m *mySQLMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange plu return sql, nil } -func (m *mySQLMacroEngine) evaluateMacro(name string, args []string) (string, error) { +func (m *mySQLMacroEngine) evaluateMacro(timeRange plugins.DataTimeRange, query plugins.DataSubQuery, name string, args []string) (string, error) { switch name { case "__timeEpoch", "__time": if len(args) == 0 { @@ -74,11 +69,11 @@ func (m *mySQLMacroEngine) evaluateMacro(name string, args []string) (string, er return "", fmt.Errorf("missing time column argument for macro %v", name) } - return fmt.Sprintf("%s BETWEEN FROM_UNIXTIME(%d) AND FROM_UNIXTIME(%d)", args[0], m.timeRange.GetFromAsSecondsEpoch(), m.timeRange.GetToAsSecondsEpoch()), nil + return fmt.Sprintf("%s BETWEEN FROM_UNIXTIME(%d) AND FROM_UNIXTIME(%d)", args[0], timeRange.GetFromAsSecondsEpoch(), timeRange.GetToAsSecondsEpoch()), nil case "__timeFrom": - return fmt.Sprintf("FROM_UNIXTIME(%d)", m.timeRange.GetFromAsSecondsEpoch()), nil + return fmt.Sprintf("FROM_UNIXTIME(%d)", timeRange.GetFromAsSecondsEpoch()), nil case "__timeTo": - return fmt.Sprintf("FROM_UNIXTIME(%d)", m.timeRange.GetToAsSecondsEpoch()), nil + return fmt.Sprintf("FROM_UNIXTIME(%d)", timeRange.GetToAsSecondsEpoch()), nil case "__timeGroup": if len(args) < 2 { return "", fmt.Errorf("macro %v needs time column and interval", name) @@ -88,14 +83,14 @@ func (m *mySQLMacroEngine) evaluateMacro(name string, args []string) (string, er return "", fmt.Errorf("error parsing interval %v", args[1]) } if len(args) == 3 { - err := sqleng.SetupFillmode(m.query, interval, args[2]) + err := sqleng.SetupFillmode(query, interval, args[2]) if err != nil { return "", err } } return fmt.Sprintf("UNIX_TIMESTAMP(%s) DIV %.0f * %.0f", args[0], interval.Seconds(), interval.Seconds()), nil case "__timeGroupAlias": - tg, err := m.evaluateMacro("__timeGroup", args) + tg, err := m.evaluateMacro(timeRange, query, "__timeGroup", args) if err == nil { return tg + " AS \"time\"", nil } @@ -104,16 +99,16 @@ func (m *mySQLMacroEngine) evaluateMacro(name string, args []string) (string, er 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], m.timeRange.GetFromAsSecondsEpoch(), args[0], m.timeRange.GetToAsSecondsEpoch()), nil + return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], timeRange.GetFromAsSecondsEpoch(), args[0], timeRange.GetToAsSecondsEpoch()), nil case "__unixEpochNanoFilter": 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], m.timeRange.GetFromAsTimeUTC().UnixNano(), args[0], m.timeRange.GetToAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], timeRange.GetFromAsTimeUTC().UnixNano(), args[0], timeRange.GetToAsTimeUTC().UnixNano()), nil case "__unixEpochNanoFrom": - return fmt.Sprintf("%d", m.timeRange.GetFromAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%d", timeRange.GetFromAsTimeUTC().UnixNano()), nil case "__unixEpochNanoTo": - return fmt.Sprintf("%d", m.timeRange.GetToAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%d", timeRange.GetToAsTimeUTC().UnixNano()), nil case "__unixEpochGroup": if len(args) < 2 { return "", fmt.Errorf("macro %v needs time column and interval and optional fill value", name) @@ -123,14 +118,14 @@ func (m *mySQLMacroEngine) evaluateMacro(name string, args []string) (string, er return "", fmt.Errorf("error parsing interval %v", args[1]) } if len(args) == 3 { - err := sqleng.SetupFillmode(m.query, interval, args[2]) + err := sqleng.SetupFillmode(query, interval, args[2]) if err != nil { return "", err } } return fmt.Sprintf("%s DIV %v * %v", args[0], interval.Seconds(), interval.Seconds()), nil case "__unixEpochGroupAlias": - tg, err := m.evaluateMacro("__unixEpochGroup", args) + tg, err := m.evaluateMacro(timeRange, query, "__unixEpochGroup", args) if err == nil { return tg + " AS \"time\"", nil } diff --git a/pkg/tsdb/mysql/macros_test.go b/pkg/tsdb/mysql/macros_test.go index 1fa57cfd115..275b35cfa78 100644 --- a/pkg/tsdb/mysql/macros_test.go +++ b/pkg/tsdb/mysql/macros_test.go @@ -3,12 +3,15 @@ package mysql import ( "fmt" "strconv" + "sync" "testing" "time" + "github.com/grafana/grafana/pkg/components/simplejson" "github.com/grafana/grafana/pkg/infra/log" "github.com/grafana/grafana/pkg/plugins" . "github.com/smartystreets/goconvey/convey" + "github.com/stretchr/testify/require" ) func TestMacroEngine(t *testing.T) { @@ -188,3 +191,33 @@ func TestMacroEngine(t *testing.T) { }) }) } + +func TestMacroEngineConcurrency(t *testing.T) { + engine := newMysqlMacroEngine(log.New("test")) + query1 := plugins.DataSubQuery{ + Model: simplejson.New(), + } + query2 := plugins.DataSubQuery{ + Model: simplejson.New(), + } + from := time.Date(2018, 4, 12, 18, 0, 0, 0, time.UTC) + to := from.Add(5 * time.Minute) + timeRange := plugins.DataTimeRange{From: "5m", To: "now", Now: to} + + var wg sync.WaitGroup + wg.Add(2) + + go func(query plugins.DataSubQuery) { + defer wg.Done() + _, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m')") + require.NoError(t, err) + }(query1) + + go func(query plugins.DataSubQuery) { + _, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m')") + require.NoError(t, err) + defer wg.Done() + }(query2) + + wg.Wait() +} diff --git a/pkg/tsdb/postgres/macros.go b/pkg/tsdb/postgres/macros.go index 5964330a8c7..c0ccc93f704 100644 --- a/pkg/tsdb/postgres/macros.go +++ b/pkg/tsdb/postgres/macros.go @@ -16,8 +16,6 @@ const sExpr = `\$` + rsIdentifier + `\(([^\)]*)\)` type postgresMacroEngine struct { *sqleng.SQLMacroEngineBase - timeRange plugins.DataTimeRange - query plugins.DataSubQuery timescaledb bool } @@ -30,8 +28,6 @@ func newPostgresMacroEngine(timescaledb bool) sqleng.SQLMacroEngine { func (m *postgresMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange plugins.DataTimeRange, sql string) (string, error) { - m.timeRange = timeRange - m.query = query // TODO: Handle error rExp, _ := regexp.Compile(sExpr) var macroError error @@ -55,7 +51,7 @@ func (m *postgresMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange for i, arg := range args { args[i] = strings.Trim(arg, " ") } - res, err := m.evaluateMacro(groups[1], args) + res, err := m.evaluateMacro(timeRange, query, groups[1], args) if err != nil && macroError == nil { macroError = err return "macro_error()" @@ -71,7 +67,7 @@ func (m *postgresMacroEngine) Interpolate(query plugins.DataSubQuery, timeRange } //nolint: gocyclo -func (m *postgresMacroEngine) evaluateMacro(name string, args []string) (string, error) { +func (m *postgresMacroEngine) evaluateMacro(timeRange plugins.DataTimeRange, query plugins.DataSubQuery, name string, args []string) (string, error) { switch name { case "__time": if len(args) == 0 { @@ -88,11 +84,11 @@ func (m *postgresMacroEngine) evaluateMacro(name string, args []string) (string, return "", fmt.Errorf("missing time column argument for macro %v", name) } - return fmt.Sprintf("%s BETWEEN '%s' AND '%s'", args[0], m.timeRange.GetFromAsTimeUTC().Format(time.RFC3339Nano), m.timeRange.GetToAsTimeUTC().Format(time.RFC3339Nano)), nil + return fmt.Sprintf("%s BETWEEN '%s' AND '%s'", args[0], timeRange.GetFromAsTimeUTC().Format(time.RFC3339Nano), timeRange.GetToAsTimeUTC().Format(time.RFC3339Nano)), nil case "__timeFrom": - return fmt.Sprintf("'%s'", m.timeRange.GetFromAsTimeUTC().Format(time.RFC3339Nano)), nil + return fmt.Sprintf("'%s'", timeRange.GetFromAsTimeUTC().Format(time.RFC3339Nano)), nil case "__timeTo": - return fmt.Sprintf("'%s'", m.timeRange.GetToAsTimeUTC().Format(time.RFC3339Nano)), nil + return fmt.Sprintf("'%s'", timeRange.GetToAsTimeUTC().Format(time.RFC3339Nano)), nil case "__timeGroup": if len(args) < 2 { return "", fmt.Errorf("macro %v needs time column and interval and optional fill value", name) @@ -102,7 +98,7 @@ func (m *postgresMacroEngine) evaluateMacro(name string, args []string) (string, return "", fmt.Errorf("error parsing interval %v", args[1]) } if len(args) == 3 { - err := sqleng.SetupFillmode(m.query, interval, args[2]) + err := sqleng.SetupFillmode(query, interval, args[2]) if err != nil { return "", err } @@ -118,7 +114,7 @@ func (m *postgresMacroEngine) evaluateMacro(name string, args []string) (string, interval.Seconds(), ), nil case "__timeGroupAlias": - tg, err := m.evaluateMacro("__timeGroup", args) + tg, err := m.evaluateMacro(timeRange, query, "__timeGroup", args) if err == nil { return tg + " AS \"time\"", nil } @@ -127,16 +123,16 @@ func (m *postgresMacroEngine) evaluateMacro(name string, args []string) (string, 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], m.timeRange.GetFromAsSecondsEpoch(), args[0], m.timeRange.GetToAsSecondsEpoch()), nil + return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], timeRange.GetFromAsSecondsEpoch(), args[0], timeRange.GetToAsSecondsEpoch()), nil case "__unixEpochNanoFilter": 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], m.timeRange.GetFromAsTimeUTC().UnixNano(), args[0], m.timeRange.GetToAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], timeRange.GetFromAsTimeUTC().UnixNano(), args[0], timeRange.GetToAsTimeUTC().UnixNano()), nil case "__unixEpochNanoFrom": - return fmt.Sprintf("%d", m.timeRange.GetFromAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%d", timeRange.GetFromAsTimeUTC().UnixNano()), nil case "__unixEpochNanoTo": - return fmt.Sprintf("%d", m.timeRange.GetToAsTimeUTC().UnixNano()), nil + return fmt.Sprintf("%d", timeRange.GetToAsTimeUTC().UnixNano()), nil case "__unixEpochGroup": if len(args) < 2 { return "", fmt.Errorf("macro %v needs time column and interval and optional fill value", name) @@ -146,14 +142,14 @@ func (m *postgresMacroEngine) evaluateMacro(name string, args []string) (string, return "", fmt.Errorf("error parsing interval %v", args[1]) } if len(args) == 3 { - err := sqleng.SetupFillmode(m.query, interval, args[2]) + err := sqleng.SetupFillmode(query, interval, args[2]) if err != nil { return "", err } } return fmt.Sprintf("floor(%s/%v)*%v", args[0], interval.Seconds(), interval.Seconds()), nil case "__unixEpochGroupAlias": - tg, err := m.evaluateMacro("__unixEpochGroup", args) + tg, err := m.evaluateMacro(timeRange, query, "__unixEpochGroup", args) if err == nil { return tg + " AS \"time\"", nil } diff --git a/pkg/tsdb/postgres/macros_test.go b/pkg/tsdb/postgres/macros_test.go index cf4696f0416..882dda43242 100644 --- a/pkg/tsdb/postgres/macros_test.go +++ b/pkg/tsdb/postgres/macros_test.go @@ -3,9 +3,11 @@ package postgres import ( "fmt" "strconv" + "sync" "testing" "time" + "github.com/grafana/grafana/pkg/components/simplejson" "github.com/grafana/grafana/pkg/plugins" "github.com/stretchr/testify/require" ) @@ -214,3 +216,33 @@ func TestMacroEngine(t *testing.T) { }) }) } + +func TestMacroEngineConcurrency(t *testing.T) { + engine := newPostgresMacroEngine(false) + query1 := plugins.DataSubQuery{ + Model: simplejson.New(), + } + query2 := plugins.DataSubQuery{ + Model: simplejson.New(), + } + from := time.Date(2018, 4, 12, 18, 0, 0, 0, time.UTC) + to := from.Add(5 * time.Minute) + timeRange := plugins.DataTimeRange{From: "5m", To: "now", Now: to} + + var wg sync.WaitGroup + wg.Add(2) + + go func(query plugins.DataSubQuery) { + defer wg.Done() + _, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m')") + require.NoError(t, err) + }(query1) + + go func(query plugins.DataSubQuery) { + _, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m')") + require.NoError(t, err) + defer wg.Done() + }(query2) + + wg.Wait() +}