diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go b/connectors/grafana-plugin/pkg/plugin/plugin.go index 4b362ba..bc256cd 100644 --- a/connectors/grafana-plugin/pkg/plugin/plugin.go +++ b/connectors/grafana-plugin/pkg/plugin/plugin.go @@ -157,6 +157,7 @@ type queryParam struct { Database string `json:"database"` Sql string `json:"sql"` Format string `json:"format"` + IntervalMS int64 `json:"-"` } type QueryDataReq struct { @@ -262,8 +263,7 @@ func (d *IoTDBDataSource) query(cxt context.Context, pCtx backend.PluginContext, return response } - qp.StartTime = query.TimeRange.From.UnixNano() / 1000000 - qp.EndTime = query.TimeRange.To.UnixNano() / 1000000 + applyQueryRuntimeValues(qp, query) if qp.SqlType == TableModelSqlType { return d.queryTableModel(cxt, qp) @@ -358,6 +358,12 @@ func (d *IoTDBDataSource) query(cxt context.Context, pCtx backend.PluginContext, return response } +func applyQueryRuntimeValues(qp *queryParam, query backend.DataQuery) { + qp.StartTime = query.TimeRange.From.UnixNano() / 1000000 + qp.EndTime = query.TimeRange.To.UnixNano() / 1000000 + qp.IntervalMS = query.Interval.Milliseconds() +} + func recoverType(m []interface{}) interface{} { if len(m) > 0 { switch m[0].(type) { diff --git a/connectors/grafana-plugin/pkg/plugin/table_query.go b/connectors/grafana-plugin/pkg/plugin/table_query.go index a53eff1..7c115ae 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_query.go +++ b/connectors/grafana-plugin/pkg/plugin/table_query.go @@ -73,8 +73,15 @@ var timeFilterRe = regexp.MustCompile(`\$__timeFilter\(\s*((?:[^()]|\([^()]*\))* var ( timeFromRe = regexp.MustCompile(`\$__timeFrom\b(?:\s*\(\s*\))?`) timeToRe = regexp.MustCompile(`\$__timeTo\b(?:\s*\(\s*\))?`) + // These patterns intentionally match the macro prefix. hasStandaloneMacro + // and replaceStandaloneMacro reject an identifier byte after the match, so + // $__interval_ms is not mistaken for $__interval. + intervalRe = regexp.MustCompile(`\$__interval`) + intervalMSRe = regexp.MustCompile(`\$__interval_ms`) ) +const invalidIntervalMacroMessage = "Grafana query interval must be positive when $__interval or $__interval_ms is used" + // formatTimeLiteral renders a panel-range bound as an ISO 8601 UTC timestamp // literal (e.g. 2020-09-13T12:26:40.000+00:00). The server parses such a // literal in its own configured timestamp precision, so the expansion works @@ -84,16 +91,37 @@ func formatTimeLiteral(ms int64) string { return time.UnixMilli(ms).UTC().Format("2006-01-02T15:04:05.000") + "+00:00" } -// expandTableMacros rewrites the Grafana time macros a dashboard author can put -// in table-model SQL into concrete bounds for the panel's range: +// expandTableMacros rewrites the Grafana time and interval macros a dashboard +// author can put in table-model SQL. // // $__timeFilter(col) -> (col >= AND col <= ) // $__timeFrom[()] -> // $__timeTo[()] -> +// $__interval -> a fixed-width IoTDB duration literal +// $__interval_ms -> the interval in milliseconds, per Grafana's contract // // Bounds are ISO 8601 UTC timestamp literals, which IoTDB compares against // TIMESTAMP columns independently of the server's timestamp precision. -func expandTableMacros(sql string, startMs int64, endMs int64) string { +func expandTableMacros(sql string, startMs int64, endMs int64, intervalMS int64) (string, error) { + hasInterval := hasStandaloneMacro(sql, intervalRe) + hasIntervalMS := hasStandaloneMacro(sql, intervalMSRe) + if (hasInterval || hasIntervalMS) && intervalMS <= 0 { + // Defensive validation for direct callers. queryTableModel performs the + // authoritative request-path check before acquiring an RPC session. + return "", errors.New(invalidIntervalMacroMessage) + } + + if hasIntervalMS { + sql = replaceStandaloneMacro(sql, intervalMSRe, strconv.FormatInt(intervalMS, 10)) + } + if hasInterval { + duration, err := formatIoTDBDuration(intervalMS) + if err != nil { + return "", err + } + sql = replaceStandaloneMacro(sql, intervalRe, duration) + } + from := formatTimeLiteral(startMs) to := formatTimeLiteral(endMs) sql = timeFilterRe.ReplaceAllStringFunc(sql, func(m string) string { @@ -105,7 +133,67 @@ func expandTableMacros(sql string, startMs int64, endMs int64) string { }) sql = timeFromRe.ReplaceAllString(sql, from) sql = timeToRe.ReplaceAllString(sql, to) - return sql + return sql, nil +} + +// hasStandaloneMacro reports whether re has a match that is not followed by an +// identifier character. Go's regexp package intentionally has no lookahead, +// so the boundary check is performed while scanning matches. +func hasStandaloneMacro(sql string, re *regexp.Regexp) bool { + for _, match := range re.FindAllStringIndex(sql, -1) { + if match[1] == len(sql) || !isSQLIdentifierByte(sql[match[1]]) { + return true + } + } + return false +} + +func replaceStandaloneMacro(sql string, re *regexp.Regexp, replacement string) string { + matches := re.FindAllStringIndex(sql, -1) + if len(matches) == 0 { + return sql + } + var b strings.Builder + last := 0 + for _, match := range matches { + if match[1] < len(sql) && isSQLIdentifierByte(sql[match[1]]) { + continue + } + b.WriteString(sql[last:match[0]]) + b.WriteString(replacement) + last = match[1] + } + b.WriteString(sql[last:]) + return b.String() +} + +func isSQLIdentifierByte(b byte) bool { + return b == '_' || b >= 'a' && b <= 'z' || b >= 'A' && b <= 'Z' || b >= '0' && b <= '9' +} + +// formatIoTDBDuration uses only fixed-width units accepted by IoTDB and +// avoids calendar month/year semantics. The largest exact unit is selected. +func formatIoTDBDuration(intervalMS int64) (string, error) { + if intervalMS <= 0 { + return "", errors.New("Grafana query interval must be positive") + } + units := []struct { + milliseconds int64 + suffix string + }{ + {7 * 24 * 60 * 60 * 1000, "w"}, + {24 * 60 * 60 * 1000, "d"}, + {60 * 60 * 1000, "h"}, + {60 * 1000, "m"}, + {1000, "s"}, + {1, "ms"}, + } + for _, unit := range units { + if intervalMS%unit.milliseconds == 0 { + return strconv.FormatInt(intervalMS/unit.milliseconds, 10) + unit.suffix, nil + } + } + return "", errors.New("cannot format Grafana query interval") } // quoteTableIdentifier wraps a table-model identifier in double quotes @@ -215,7 +303,12 @@ func (d *IoTDBDataSource) getTablePool() (*client.TableSessionPool, error) { func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) backend.DataResponse { response := backend.DataResponse{} - sql := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime) + if (hasStandaloneMacro(qp.Sql, intervalRe) || hasStandaloneMacro(qp.Sql, intervalMSRe)) && qp.IntervalMS <= 0 { + // This is the authoritative guard: reject invalid Grafana input before + // getTablePool can create or acquire an RPC session. + response.Error = errors.New(invalidIntervalMacroMessage) + return response + } pool, err := d.getTablePool() if err != nil { @@ -247,6 +340,11 @@ func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) b timeout = ms } } + sql, err := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime, qp.IntervalMS) + if err != nil { + response.Error = err + return response + } resultSet, err := session.ExecuteQueryStatement(sql, &timeout) if err != nil { response.Error = err diff --git a/connectors/grafana-plugin/pkg/plugin/table_query_test.go b/connectors/grafana-plugin/pkg/plugin/table_query_test.go index 899bc14..29c65eb 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_query_test.go +++ b/connectors/grafana-plugin/pkg/plugin/table_query_test.go @@ -18,7 +18,10 @@ package plugin import ( + "context" + "encoding/json" "errors" + "strings" "testing" "time" @@ -80,7 +83,10 @@ func TestExpandTableMacros(t *testing.T) { for _, c := range cases { t.Run(c.name, func(t *testing.T) { - got := expandTableMacros(c.in, from, to) + got, err := expandTableMacros(c.in, from, to, 0) + if err != nil { + t.Fatalf("expandTableMacros() unexpected error: %v", err) + } if got != c.want { t.Fatalf("expandTableMacros() = %q, want %q", got, c.want) } @@ -88,6 +94,141 @@ func TestExpandTableMacros(t *testing.T) { } } +func TestExpandTableIntervalMacros(t *testing.T) { + const from int64 = 1600000000000 + const to int64 = 1600000001000 + + cases := []struct { + name string + sql string + interval int64 + want string + wantErr string + }{ + { + name: "expands duration and milliseconds together", + sql: "SELECT date_bin($__interval, time) + $__interval_ms AS bucket_time FROM table1", + interval: 120000, + want: "SELECT date_bin(2m, time) + 120000 AS bucket_time FROM table1", + }, + {name: "separate interval macros do not overlap", sql: "SELECT $__interval, $__interval_ms", interval: 120000, want: "SELECT 2m, 120000"}, + {name: "milliseconds", sql: "SELECT $__interval", interval: 500, want: "SELECT 500ms"}, + {name: "seconds", sql: "SELECT $__interval", interval: 1000, want: "SELECT 1s"}, + {name: "minutes", sql: "SELECT $__interval", interval: 120000, want: "SELECT 2m"}, + {name: "hours", sql: "SELECT $__interval", interval: 3600000, want: "SELECT 1h"}, + {name: "days", sql: "SELECT $__interval", interval: 86400000, want: "SELECT 1d"}, + {name: "weeks", sql: "SELECT $__interval", interval: 604800000, want: "SELECT 1w"}, + {name: "non exact duration uses milliseconds", sql: "SELECT $__interval", interval: 1500, want: "SELECT 1500ms"}, + {name: "Grafana interval milliseconds contract", sql: "SELECT $__interval_ms", interval: 120000, want: "SELECT 120000"}, + {name: "identifier boundaries are preserved", sql: "SELECT $__intervalish, $__interval_ms_extra", interval: 120000, want: "SELECT $__intervalish, $__interval_ms_extra"}, + {name: "interval is ignored when no interval macro exists", sql: "SELECT $__timeFrom", interval: 0, want: "SELECT 2020-09-13T12:26:40.000+00:00"}, + {name: "zero interval fails", sql: "SELECT $__interval", interval: 0, wantErr: "Grafana query interval must be positive"}, + {name: "negative interval fails", sql: "SELECT $__interval_ms", interval: -1, wantErr: "Grafana query interval must be positive"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := expandTableMacros(tc.sql, from, to, tc.interval) + if tc.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantErr) { + t.Fatalf("expandTableMacros() error = %v, want substring %q", err, tc.wantErr) + } + return + } + if err != nil { + t.Fatalf("expandTableMacros() unexpected error: %v", err) + } + if got != tc.want { + t.Fatalf("expandTableMacros() = %q, want %q", got, tc.want) + } + }) + } +} + +func TestExpandTableIntervalMacrosWithExplicitOrigin(t *testing.T) { + cases := []struct { + name string + from int64 + to int64 + sql string + want string + }{ + { + name: "date bin origin offset by thirty seconds", + from: 1600000030000, + to: 1600000150000, + sql: "SELECT date_bin($__interval, time, $__timeFrom) AS bucket_time FROM table1", + want: "SELECT date_bin(2m, time, 2020-09-13T12:27:10.000+00:00) AS bucket_time FROM table1", + }, + { + name: "hop origin offset by forty five seconds", + from: 1600000045000, + to: 1600000165000, + sql: "SELECT * FROM HOP(DATA => table1, SLIDE => $__interval, SIZE => 1m, ORIGIN => $__timeFrom)", + want: "SELECT * FROM HOP(DATA => table1, SLIDE => 2m, SIZE => 1m, ORIGIN => 2020-09-13T12:27:25.000+00:00)", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := expandTableMacros(tc.sql, tc.from, tc.to, 120000) + if err != nil { + t.Fatalf("expandTableMacros() unexpected error: %v", err) + } + if got != tc.want { + t.Fatalf("expandTableMacros() = %q, want %q", got, tc.want) + } + }) + } +} + +func TestQueryParamRuntimeIntervalOverridesJSON(t *testing.T) { + qp, msg := verifyQuery(backend.DataQuery{JSON: []byte(`{"sqlType":"SQL: Table Model","sql":"SELECT $__interval_ms","database":"db1","intervalMS":999}`)}) + if msg != "" { + t.Fatalf("valid table query rejected: %q", msg) + } + if qp.IntervalMS != 0 { + t.Fatalf("JSON intervalMS should not populate runtime field: %d", qp.IntervalMS) + } + serialized, err := json.Marshal(qp) + if err != nil { + t.Fatalf("marshal query param: %v", err) + } + if strings.Contains(string(serialized), "intervalMS") { + t.Fatalf("runtime intervalMS must not be serialized: %s", serialized) + } + query := backend.DataQuery{ + Interval: 120 * time.Second, + TimeRange: backend.TimeRange{ + From: ts(1600000000000), + To: ts(1600000001000), + }, + } + applyQueryRuntimeValues(qp, query) + if qp.IntervalMS != 120000 || qp.StartTime != 1600000000000 || qp.EndTime != 1600000001000 { + t.Fatalf("runtime values = start %d, end %d, interval %d", qp.StartTime, qp.EndTime, qp.IntervalMS) + } +} + +func TestQueryTableModelRejectsNonPositiveIntervalBeforeRPC(t *testing.T) { + _, expandErr := expandTableMacros("SELECT $__interval", 0, 0, 0) + if expandErr == nil || expandErr.Error() != invalidIntervalMacroMessage { + t.Fatalf("expandTableMacros() error = %v, want %q", expandErr, invalidIntervalMacroMessage) + } + + d := &IoTDBDataSource{Ulr: "http://invalid-host:18080"} + response := d.queryTableModel(context.Background(), &queryParam{ + Sql: "SELECT $__interval FROM table1", + Database: "db1", + IntervalMS: 0, + }) + if response.Error == nil || response.Error.Error() != invalidIntervalMacroMessage { + t.Fatalf("queryTableModel() error = %v, want early positive-interval error", response.Error) + } + if d.tablePool != nil { + t.Fatalf("invalid interval should be rejected before creating an RPC pool") + } +} + func TestQuoteTableIdentifier(t *testing.T) { if got := quoteTableIdentifier("test"); got != `"test"` { t.Fatalf("plain identifier = %q", got)