diff --git a/connectors/grafana-plugin/README.md b/connectors/grafana-plugin/README.md index 974916bc..b3805011 100644 --- a/connectors/grafana-plugin/README.md +++ b/connectors/grafana-plugin/README.md @@ -69,7 +69,7 @@ Click the `New Dashboard` icon on the top right, and select `Add an empty panel` -Grafana plugin supports SQL: Full Customized mode and SQL: Drop-down List mode, and the default mode is SQL: Full Customized mode. +Grafana plugin supports SQL: Full Customized mode and SQL: Drop-down List mode for the tree model, and SQL: Table Model mode for the table model. The default mode is SQL: Full Customized mode. @@ -119,6 +119,37 @@ Select a time series in the TIME-SERIES selection box, select a function in the +##### SQL: Table Model + +Queries IoTDB's table model (IoTDB 2.x) through the native Go client ([apache/iotdb-client-go](https://github.com/apache/iotdb-client-go)). Enter a standard SQL statement in the SQL input box, for example: + +```sql +SELECT time, device_id, temperature FROM table1 WHERE $__timeFilter(time) +``` + +Unlike the tree-model modes, which go through the REST service, this mode connects to the IoTDB RPC port (6667 by default). The data source's `rpc address` option sets that endpoint explicitly; when it is left empty, the URL's host with port 6667 is used. + +DATABASE input box: the database to run the statement against. Required — every query first runs `USE ` on its pooled session, so results stay deterministic regardless of which session the pool hands out. Fully-qualified `database.table` references are still allowed and take precedence over the session database. + +SQL input box: a single table-model SELECT statement. The following macros are expanded by the plugin before the statement is sent: + +| Macro | Expands to | +| ----- | ---------- | +| `$__timeFilter(col)` | `(col >= AND col <= )`; `col` defaults to `time` when omitted | +| `$__timeFrom`, `$__timeFrom()` | the panel range start as an ISO 8601 UTC timestamp literal | +| `$__timeTo`, `$__timeTo()` | the panel range end as an ISO 8601 UTC timestamp literal | + +The macros expand to ISO 8601 timestamp literals (e.g. `2020-09-13T12:26:40.000+00:00`), which the server interprets in its own configured `timestamp_precision` — so time filtering works unchanged on `ms`, `us` and `ns` servers, and TIMESTAMP values are likewise converted by the client using the server-reported precision. + +FORMAT option: + +* `Time series` (default): rows are sorted by the first TIMESTAMP column, and a result that contains tag/string columns next to numeric columns is pivoted into one series per tag combination — so a query returning several devices renders as separate lines in a time-series panel. Note that the pivot requires timestamps without nulls; results are sorted by the plugin, so an `ORDER BY` clause is not needed. +* `Table`: rows are returned exactly as the server sent them (use this with the table panel, or when your `ORDER BY` should be preserved). + +Result columns are typed from the result set's data types: TIMESTAMP columns become Grafana time fields, INT32/INT64 become integers, FLOAT/DOUBLE become floats, BOOLEAN becomes booleans, and TEXT/STRING/DATE/BLOB render as strings. + +There is no server-imposed row cap on this path; add a `LIMIT` clause when querying wide time ranges over dense data to bound the result size. + #### Support for variables and template functions Both SQL: Full Customized and SQL: Drop-down List input methods support the variable and template functions of grafana. In the following example, raw input method is used, and aggregation is similar. diff --git a/connectors/grafana-plugin/go.mod b/connectors/grafana-plugin/go.mod index 8f7ca6be..fbc09514 100644 --- a/connectors/grafana-plugin/go.mod +++ b/connectors/grafana-plugin/go.mod @@ -19,11 +19,15 @@ go 1.21 toolchain go1.21.0 -require github.com/grafana/grafana-plugin-sdk-go v0.250.0 +require ( + github.com/apache/iotdb-client-go/v2 v2.0.8 + github.com/grafana/grafana-plugin-sdk-go v0.250.0 +) require ( github.com/BurntSushi/toml v1.3.2 // indirect github.com/apache/arrow/go/v15 v15.0.2 // indirect + github.com/apache/thrift v0.17.0 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go b/connectors/grafana-plugin/pkg/plugin/plugin.go index a65e987b..4b362bab 100644 --- a/connectors/grafana-plugin/pkg/plugin/plugin.go +++ b/connectors/grafana-plugin/pkg/plugin/plugin.go @@ -28,8 +28,10 @@ import ( "net/http" "strconv" "strings" + "sync" "time" + "github.com/apache/iotdb-client-go/v2/client" "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/backend/httpclient" "github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt" @@ -70,7 +72,8 @@ func ApacheIoTDBDatasource(ctx context.Context, d backend.DataSourceInstanceSett if password, exists := d.DecryptedSecureJSONData["password"]; exists { authorization = "Basic " + base64.StdEncoding.EncodeToString([]byte(dm.Username+":"+password)) } - return &IoTDBDataSource{CallResourceHandler: iotdbResourceHandler(authorization, httpClient), Username: dm.Username, Ulr: dm.Url, httpClient: httpClient}, nil + password := d.DecryptedSecureJSONData["password"] + return &IoTDBDataSource{CallResourceHandler: iotdbResourceHandler(authorization, httpClient), Username: dm.Username, Ulr: dm.Url, RPCAddress: dm.RPCAddress, password: password, httpClient: httpClient}, nil } // SampleDatasource is an example datasource which can respond to data queries, reports @@ -79,7 +82,14 @@ type IoTDBDataSource struct { backend.CallResourceHandler Username string Ulr string + RPCAddress string + password string httpClient *http.Client + + // Native-client session pool for the table-model mode, created lazily by + // getTablePool on the first table query. + tablePoolMu sync.Mutex + tablePool *client.TableSessionPool } // Dispose here tells plugin SDK that plugin wants to clean up resources when a new instance @@ -88,6 +98,12 @@ type IoTDBDataSource struct { func (d *IoTDBDataSource) Dispose() { // Clean up datasource instance resources. d.httpClient.CloseIdleConnections() + d.tablePoolMu.Lock() + defer d.tablePoolMu.Unlock() + if d.tablePool != nil { + d.tablePool.Close() + d.tablePool = nil + } } // QueryData handles multiple queries and returns multiple responses. @@ -113,6 +129,10 @@ func (d *IoTDBDataSource) QueryData(ctx context.Context, req *backend.QueryDataR type dataSourceModel struct { Username string `json:"username"` Url string `json:"url"` + // RPCAddress is the IoTDB RPC endpoint (host or host:port) the native + // client used by the table-model mode connects to. When empty, the URL's + // host with the default RPC port 6667 is used. + RPCAddress string `json:"rpcAddress"` } type groupBy struct { @@ -134,6 +154,9 @@ type queryParam struct { FillClauses string `json:"fillClauses"` GroupBy groupBy `json:"groupBy"` Hide bool `json:"hide"` + Database string `json:"database"` + Sql string `json:"sql"` + Format string `json:"format"` } type QueryDataReq struct { @@ -200,6 +223,17 @@ func verifyQuery(query backend.DataQuery) (qp *queryParam, errMsg string) { return nil, "Input error, FROM is required" } } + } else if qp.SqlType == TableModelSqlType { + if strings.TrimSpace(qp.Sql) == "" { + return nil, "Input error, SQL is required" + } + // The database is required: every query then USEs it on its pooled + // session, so a session's leftover USE state from a previous query + // can never influence the result (PooledTableSession.Close only + // resets the database when the pool has one configured). + if strings.TrimSpace(qp.Database) == "" { + return nil, "Input error, DATABASE is required" + } } else { return nil, "none" } @@ -231,6 +265,10 @@ func (d *IoTDBDataSource) query(cxt context.Context, pCtx backend.PluginContext, qp.StartTime = query.TimeRange.From.UnixNano() / 1000000 qp.EndTime = query.TimeRange.To.UnixNano() / 1000000 + if qp.SqlType == TableModelSqlType { + return d.queryTableModel(cxt, qp) + } + if qp.SqlType == "SQL: Drop-down List" { qp.Control = "" var expressions []string = qp.Paths[len(qp.Paths)-1:] diff --git a/connectors/grafana-plugin/pkg/plugin/table_query.go b/connectors/grafana-plugin/pkg/plugin/table_query.go new file mode 100644 index 00000000..a53eff1b --- /dev/null +++ b/connectors/grafana-plugin/pkg/plugin/table_query.go @@ -0,0 +1,474 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package plugin + +import ( + "context" + "encoding/hex" + "errors" + "fmt" + "net" + "net/url" + "regexp" + "sort" + "strconv" + "strings" + "time" + + "github.com/apache/iotdb-client-go/v2/client" + "github.com/grafana/grafana-plugin-sdk-go/backend" + "github.com/grafana/grafana-plugin-sdk-go/backend/log" + "github.com/grafana/grafana-plugin-sdk-go/data" +) + +// TableModelSqlType is the QueryEditor mode that runs standard table-model SQL +// through the native Go client (apache/iotdb-client-go), as opposed to the +// tree-model modes ("SQL: Full Customized" / "SQL: Drop-down List") that build +// root.* paths and go through the REST /grafana endpoints. +const TableModelSqlType = "SQL: Table Model" + +// Result formats for the table-model mode. Time series (the default) sorts +// rows ascending by the first TIMESTAMP column and pivots a long-shaped result +// (time + tag columns + value columns) into one labeled series per tag +// combination; Table returns the rows exactly as the server sent them. +const ( + tableFormatTimeSeries = "Time series" + tableFormatTable = "Table" +) + +// Connection settings for the native client. The RPC endpoint is the +// datasource's "rpc address" option, or the URL's host with the default RPC +// port when unset. The pool is created lazily on the first table-model query +// so datasources that only use the tree model never open an RPC connection. +const ( + defaultRPCPort = "6667" + tablePoolMaxSize = 8 + tableConnectTimeoutMs = 10000 + tableWaitSessionTimeoutMs = 60000 + defaultQueryTimeoutMs = int64(60000) +) + +// timeFilterRe matches Grafana's $__timeFilter(column) macro; the column is +// optional and defaults to "time". One level of nested parentheses is allowed +// so expressions like $__timeFilter(cast(x)) survive intact. +var timeFilterRe = regexp.MustCompile(`\$__timeFilter\(\s*((?:[^()]|\([^()]*\))*?)\s*\)`) + +// timeFromRe / timeToRe match $__timeFrom / $__timeTo, with or without the +// trailing () that Grafana's SQL data sources use ($__timeFrom()). +var ( + timeFromRe = regexp.MustCompile(`\$__timeFrom\b(?:\s*\(\s*\))?`) + timeToRe = regexp.MustCompile(`\$__timeTo\b(?:\s*\(\s*\))?`) +) + +// 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 +// unchanged on ms, us and ns servers — unlike a bare epoch integer, which the +// server would interpret in raw server units. +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: +// +// $__timeFilter(col) -> (col >= AND col <= ) +// $__timeFrom[()] -> +// $__timeTo[()] -> +// +// 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 { + from := formatTimeLiteral(startMs) + to := formatTimeLiteral(endMs) + sql = timeFilterRe.ReplaceAllStringFunc(sql, func(m string) string { + col := strings.TrimSpace(timeFilterRe.FindStringSubmatch(m)[1]) + if col == "" { + col = "time" + } + return "(" + col + " >= " + from + " AND " + col + " <= " + to + ")" + }) + sql = timeFromRe.ReplaceAllString(sql, from) + sql = timeToRe.ReplaceAllString(sql, to) + return sql +} + +// quoteTableIdentifier wraps a table-model identifier in double quotes +// (doubling any embedded quote), the relational grammar's quoted-identifier +// form, so a database name survives the USE statement verbatim. +func quoteTableIdentifier(name string) string { + return `"` + strings.ReplaceAll(name, `"`, `""`) + `"` +} + +// tableQueryDataSet is the plugin-internal carrier for a fetched table-model +// result: column names and IoTDB type names as reported by the client, plus +// row-major cell values (values[row][col]) holding the client's native Go +// representations (time.Time for TIMESTAMP/DATE, string for TEXT/STRING, +// []byte for BLOB, bool/int32/int64/float32/float64 for the scalars, nil for +// null). +type tableQueryDataSet struct { + ColumnNames []string + DataTypes []string + Values [][]interface{} +} + +// tableResultSet is the slice of the native client's SessionDataSet the fetch +// path needs; narrowing it to an interface keeps fetchTableDataSet testable +// without a live server. Column indexes are 1-based, matching the client. +type tableResultSet interface { + Next() (bool, error) + GetColumnNames() []string + GetColumnTypes() []string + GetObjectByIndex(columnIndex int32) (interface{}, error) + Close() error +} + +// fetchTableDataSet drains a result set into a tableQueryDataSet. The caller +// owns closing the result set. +func fetchTableDataSet(rs tableResultSet) (*tableQueryDataSet, error) { + names := rs.GetColumnNames() + types := rs.GetColumnTypes() + dataSet := &tableQueryDataSet{ColumnNames: names, DataTypes: types, Values: [][]interface{}{}} + for { + hasNext, err := rs.Next() + if err != nil { + return nil, err + } + if !hasNext { + return dataSet, nil + } + row := make([]interface{}, len(names)) + for i := range names { + value, err := rs.GetObjectByIndex(int32(i) + 1) + if err != nil { + return nil, err + } + row[i] = value + } + dataSet.Values = append(dataSet.Values, row) + } +} + +// tableRPCEndpoint resolves the host and port of the IoTDB RPC service the +// native client connects to: the datasource's "rpc address" option when set +// (host or host:port), otherwise the datasource URL's host with the default +// RPC port. +func (d *IoTDBDataSource) tableRPCEndpoint() (string, string, error) { + if addr := strings.TrimSpace(d.RPCAddress); addr != "" { + if host, port, err := net.SplitHostPort(addr); err == nil { + return host, port, nil + } + return strings.Trim(addr, "[]"), defaultRPCPort, nil + } + raw := strings.TrimSpace(d.Ulr) + if u, err := url.Parse(raw); err == nil && u.Hostname() != "" { + return u.Hostname(), defaultRPCPort, nil + } + // A bare host[:restPort] without a scheme parses as opaque; retry with one. + if u, err := url.Parse("http://" + raw); err == nil && u.Hostname() != "" { + return u.Hostname(), defaultRPCPort, nil + } + return "", "", errors.New("cannot derive the IoTDB RPC host from the datasource URL; please set the rpc address option") +} + +// getTablePool lazily creates the shared native-client session pool for this +// datasource instance. Pool construction does not connect; connection errors +// surface on GetSession. +func (d *IoTDBDataSource) getTablePool() (*client.TableSessionPool, error) { + d.tablePoolMu.Lock() + defer d.tablePoolMu.Unlock() + if d.tablePool != nil { + return d.tablePool, nil + } + host, port, err := d.tableRPCEndpoint() + if err != nil { + return nil, err + } + poolConfig := &client.PoolConfig{ + Host: host, + Port: port, + UserName: d.Username, + Password: d.password, + } + pool := client.NewTableSessionPool(poolConfig, tablePoolMaxSize, tableConnectTimeoutMs, tableWaitSessionTimeoutMs, false) + d.tablePool = &pool + return d.tablePool, nil +} + +// queryTableModel runs a table-model SQL query through the native client and +// turns the result set into a Grafana data frame. +func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) backend.DataResponse { + response := backend.DataResponse{} + + sql := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime) + + pool, err := d.getTablePool() + if err != nil { + response.Error = err + return response + } + session, err := pool.GetSession() + if err != nil { + response.Error = fmt.Errorf("cannot connect to the IoTDB RPC service: %w", err) + log.DefaultLogger.Error("Cannot connect to the IoTDB RPC service", "err", err) + return response + } + defer func() { + if closeErr := session.Close(); closeErr != nil { + log.DefaultLogger.Error("Failed to return session to the pool", "err", closeErr) + } + }() + + if database := strings.TrimSpace(qp.Database); database != "" { + if err := session.ExecuteNonQueryStatement("USE " + quoteTableIdentifier(database)); err != nil { + response.Error = err + return response + } + } + + timeout := defaultQueryTimeoutMs + if deadline, ok := ctx.Deadline(); ok { + if ms := time.Until(deadline).Milliseconds(); ms > 0 { + timeout = ms + } + } + resultSet, err := session.ExecuteQueryStatement(sql, &timeout) + if err != nil { + response.Error = err + return response + } + defer func() { + if closeErr := resultSet.Close(); closeErr != nil { + log.DefaultLogger.Error("Failed to close the result set", "err", closeErr) + } + }() + + dataSet, err := fetchTableDataSet(resultSet) + if err != nil { + response.Error = err + return response + } + + response.Frames = append(response.Frames, buildTableResponseFrame(dataSet, qp.Format)) + return response +} + +// buildTableResponseFrame turns a fetched dataset into the response frame, +// honoring the query's format. In the default Time series format the rows are +// sorted ascending by the first TIMESTAMP column and a long-shaped result +// (time + string tag columns + value columns) is pivoted into one labeled +// series per tag combination, so a multi-device query renders as separate +// lines instead of one interleaved series; when the pivot does not apply (or +// fails, e.g. on a null timestamp) the plain frame is returned. The Table +// format preserves the server's row order untouched. +func buildTableResponseFrame(dataSet *tableQueryDataSet, format string) *data.Frame { + // Anything that is not explicitly the Table format gets the default + // time-series treatment, including queries saved before FORMAT existed. + isTimeSeries := !strings.EqualFold(format, tableFormatTable) + if isTimeSeries { + sortRowsByFirstTimestamp(dataSet) + } + frame := buildTableFrame(dataSet) + if isTimeSeries && frame.TimeSeriesSchema().Type == data.TimeSeriesTypeLong { + if wide, err := data.LongToWide(frame, nil); err == nil { + frame = wide + } + } + return frame +} + +// sortRowsByFirstTimestamp stably sorts the row-major values ascending by the +// first TIMESTAMP column, which both time-series rendering and the long-to-wide +// pivot require. Rows whose time cell is null sort last. +func sortRowsByFirstTimestamp(dataSet *tableQueryDataSet) { + timeCol := -1 + for i, t := range dataSet.DataTypes { + if strings.EqualFold(t, "TIMESTAMP") { + timeCol = i + break + } + } + if timeCol < 0 { + return + } + sort.SliceStable(dataSet.Values, func(a, b int) bool { + va, okA := rowTimeAt(dataSet.Values[a], timeCol) + vb, okB := rowTimeAt(dataSet.Values[b], timeCol) + if okA != okB { + return okA + } + return okA && va.Before(vb) + }) +} + +func rowTimeAt(row []interface{}, col int) (time.Time, bool) { + if col >= len(row) { + return time.Time{}, false + } + t, ok := row[col].(time.Time) + return t, ok +} + +// buildTableFrame converts the row-major dataset into a Grafana frame, one +// typed field per column driven by the reported data types; each field gathers +// a single column across all rows, so every field ends up the same length (the +// row count) that Grafana requires. +func buildTableFrame(dataSet *tableQueryDataSet) *data.Frame { + frame := data.NewFrame("response") + rowCount := len(dataSet.Values) + for col := 0; col < len(dataSet.ColumnNames); col++ { + name := dataSet.ColumnNames[col] + dataType := "" + if col < len(dataSet.DataTypes) { + dataType = dataSet.DataTypes[col] + } + columnValues := make([]interface{}, rowCount) + for row := 0; row < rowCount; row++ { + if col < len(dataSet.Values[row]) { + columnValues[row] = dataSet.Values[row][col] + } + } + frame.Fields = append(frame.Fields, buildTableField(name, dataType, columnValues)) + } + return frame +} + +// buildTableField turns one column's native client values into a typed, +// nullable Grafana field selected by the IoTDB data type. The client already +// converts TIMESTAMP (and DATE) values to time.Time using the server-reported +// timestamp precision, so no unit handling happens here. DATE and BLOB render +// as strings (yyyy-MM-dd and 0x-prefixed hex), matching the REST behavior the +// mode previously had. +func buildTableField(name string, dataType string, values []interface{}) *data.Field { + switch strings.ToUpper(dataType) { + case "TIMESTAMP": + out := make([]*time.Time, len(values)) + for i, v := range values { + if t, ok := v.(time.Time); ok { + value := t + out[i] = &value + } + } + return data.NewField(name, nil, out) + case "INT32", "INT64": + out := make([]*int64, len(values)) + for i, v := range values { + if n, ok := toInt64(v); ok { + value := n + out[i] = &value + } + } + return data.NewField(name, nil, out) + case "FLOAT", "DOUBLE": + out := make([]*float64, len(values)) + for i, v := range values { + if f, ok := toFloat64(v); ok { + value := f + out[i] = &value + } + } + return data.NewField(name, nil, out) + case "BOOLEAN": + out := make([]*bool, len(values)) + for i, v := range values { + if b, ok := v.(bool); ok { + value := b + out[i] = &value + } + } + return data.NewField(name, nil, out) + case "DATE": + out := make([]*string, len(values)) + for i, v := range values { + if t, ok := v.(time.Time); ok { + value := t.Format("2006-01-02") + out[i] = &value + } + } + return data.NewField(name, nil, out) + case "BLOB": + out := make([]*string, len(values)) + for i, v := range values { + if b, ok := v.([]byte); ok { + value := "0x" + hex.EncodeToString(b) + out[i] = &value + } + } + return data.NewField(name, nil, out) + default: + // TEXT, STRING and anything unrecognised render as strings. + out := make([]*string, len(values)) + for i, v := range values { + if v == nil { + continue + } + value := toString(v) + out[i] = &value + } + return data.NewField(name, nil, out) + } +} + +// toInt64 coerces the client's integer representations (INT32 -> int32, +// INT64 -> int64) to int64. +func toInt64(v interface{}) (int64, bool) { + switch n := v.(type) { + case int64: + return n, true + case int32: + return int64(n), true + } + return 0, false +} + +// toFloat64 coerces the client's floating representations (FLOAT -> float32, +// DOUBLE -> float64) to float64. +func toFloat64(v interface{}) (float64, bool) { + switch n := v.(type) { + case float64: + return n, true + case float32: + return float64(n), true + } + return 0, false +} + +// toString renders a value for a text column without losing it, whatever the +// client handed over. +func toString(v interface{}) string { + switch value := v.(type) { + case string: + return value + case []byte: + return "0x" + hex.EncodeToString(value) + case time.Time: + return value.UTC().Format(time.RFC3339Nano) + case bool: + return strconv.FormatBool(value) + case int32: + return strconv.FormatInt(int64(value), 10) + case int64: + return strconv.FormatInt(value, 10) + case float32: + return strconv.FormatFloat(float64(value), 'f', -1, 32) + case float64: + return strconv.FormatFloat(value, 'f', -1, 64) + default: + return fmt.Sprintf("%v", v) + } +} diff --git a/connectors/grafana-plugin/pkg/plugin/table_query_test.go b/connectors/grafana-plugin/pkg/plugin/table_query_test.go new file mode 100644 index 00000000..899bc14d --- /dev/null +++ b/connectors/grafana-plugin/pkg/plugin/table_query_test.go @@ -0,0 +1,422 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package plugin + +import ( + "errors" + "testing" + "time" + + "github.com/grafana/grafana-plugin-sdk-go/backend" + "github.com/grafana/grafana-plugin-sdk-go/data" +) + +func ts(ms int64) time.Time { + return time.UnixMilli(ms).UTC() +} + +func TestExpandTableMacros(t *testing.T) { + const from int64 = 1600000000000 // 2020-09-13T12:26:40.000+00:00 + const to int64 = 1600000001000 // 2020-09-13T12:26:41.000+00:00 + const fromLit = "2020-09-13T12:26:40.000+00:00" + const toLit = "2020-09-13T12:26:41.000+00:00" + + cases := []struct { + name string + in string + want string + }{ + { + name: "timeFilter with explicit column", + in: "SELECT time, s0 FROM db.t WHERE $__timeFilter(time)", + want: "SELECT time, s0 FROM db.t WHERE (time >= " + fromLit + " AND time <= " + toLit + ")", + }, + { + name: "timeFilter defaults to time column when empty", + in: "SELECT * FROM db.t WHERE $__timeFilter()", + want: "SELECT * FROM db.t WHERE (time >= " + fromLit + " AND time <= " + toLit + ")", + }, + { + name: "timeFrom and timeTo", + in: "SELECT * FROM db.t WHERE time >= $__timeFrom AND time <= $__timeTo", + want: "SELECT * FROM db.t WHERE time >= " + fromLit + " AND time <= " + toLit, + }, + { + name: "function form timeFrom and timeTo", + in: "SELECT * FROM db.t WHERE time >= $__timeFrom() AND time <= $__timeTo( )", + want: "SELECT * FROM db.t WHERE time >= " + fromLit + " AND time <= " + toLit, + }, + { + name: "timeFilter with one nested paren level", + in: "SELECT * FROM db.t WHERE $__timeFilter(cast(x))", + want: "SELECT * FROM db.t WHERE (cast(x) >= " + fromLit + " AND cast(x) <= " + toLit + ")", + }, + { + name: "longer identifier is not mangled", + in: "SELECT $__timeFromage FROM db.t", + want: "SELECT $__timeFromage FROM db.t", + }, + { + name: "no macros is unchanged", + in: "SELECT * FROM db.t", + want: "SELECT * FROM db.t", + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := expandTableMacros(c.in, from, to) + if got != c.want { + t.Fatalf("expandTableMacros() = %q, want %q", got, c.want) + } + }) + } +} + +func TestQuoteTableIdentifier(t *testing.T) { + if got := quoteTableIdentifier("test"); got != `"test"` { + t.Fatalf("plain identifier = %q", got) + } + if got := quoteTableIdentifier(`we"ird`); got != `"we""ird"` { + t.Fatalf("embedded quote = %q", got) + } +} + +// fakeResultSet implements tableResultSet over an in-memory row list, standing +// in for the native client's SessionDataSet (1-based column indexes). +type fakeResultSet struct { + names []string + types []string + rows [][]interface{} + cursor int + err error + closed bool +} + +func (f *fakeResultSet) Next() (bool, error) { + if f.err != nil { + return false, f.err + } + if f.cursor >= len(f.rows) { + return false, nil + } + f.cursor++ + return true, nil +} + +func (f *fakeResultSet) GetColumnNames() []string { return f.names } +func (f *fakeResultSet) GetColumnTypes() []string { return f.types } + +func (f *fakeResultSet) GetObjectByIndex(columnIndex int32) (interface{}, error) { + return f.rows[f.cursor-1][columnIndex-1], nil +} + +func (f *fakeResultSet) Close() error { + f.closed = true + return nil +} + +func TestFetchTableDataSet(t *testing.T) { + rs := &fakeResultSet{ + names: []string{"time", "device", "value"}, + types: []string{"TIMESTAMP", "STRING", "DOUBLE"}, + rows: [][]interface{}{ + {ts(1000), "d1", float64(1.5)}, + {ts(2000), nil, float64(2.5)}, // null cell passes through + }, + } + dataSet, err := fetchTableDataSet(rs) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(dataSet.Values) != 2 || len(dataSet.ColumnNames) != 3 { + t.Fatalf("unexpected shape: %d rows, %d columns", len(dataSet.Values), len(dataSet.ColumnNames)) + } + if dataSet.Values[0][1] != "d1" || dataSet.Values[1][1] != nil { + t.Fatalf("cells not carried over: %#v", dataSet.Values) + } + if tv, ok := dataSet.Values[1][0].(time.Time); !ok || !tv.Equal(ts(2000)) { + t.Fatalf("time cell = %#v, want %v", dataSet.Values[1][0], ts(2000)) + } +} + +func TestFetchTableDataSetPropagatesError(t *testing.T) { + rs := &fakeResultSet{names: []string{"a"}, types: []string{"INT64"}, err: errors.New("broken pipe")} + if _, err := fetchTableDataSet(rs); err == nil { + t.Fatalf("expected the iteration error to propagate") + } +} + +// TestBuildTableFrameRowOrientation pins the fetch orientation contract: the +// dataset rows are row-major (values[row][col]), so a field must gather a +// single column across every row, with the client's native Go value types. +func TestBuildTableFrameRowOrientation(t *testing.T) { + ds := &tableQueryDataSet{ + ColumnNames: []string{"time", "i", "d", "b", "s"}, + DataTypes: []string{"TIMESTAMP", "INT64", "DOUBLE", "BOOLEAN", "TEXT"}, + Values: [][]interface{}{ + {ts(1600000000000), int64(42), float64(3.5), true, "hello"}, + {ts(1600000001000), int64(43), float64(4.5), false, "world"}, + }, + } + + frame := buildTableFrame(ds) + if len(frame.Fields) != 5 { + t.Fatalf("expected 5 fields, got %d", len(frame.Fields)) + } + for _, f := range frame.Fields { + if f.Len() != 2 { + t.Fatalf("field %q has len %d, want 2 (row count)", f.Name, f.Len()) + } + } + + if v, ok := frame.Fields[0].At(1).(*time.Time); !ok || v == nil || !v.Equal(ts(1600000001000)) { + t.Fatalf("time[1] = %#v", frame.Fields[0].At(1)) + } + if v := frame.Fields[1].At(0).(*int64); *v != 42 { + t.Fatalf("i[0] = %d, want 42", *v) + } + if v := frame.Fields[2].At(1).(*float64); *v != 4.5 { + t.Fatalf("d[1] = %v, want 4.5", *v) + } + if v := frame.Fields[3].At(1).(*bool); *v != false { + t.Fatalf("b[1] = %v, want false", *v) + } + if v := frame.Fields[4].At(1).(*string); *v != "world" { + t.Fatalf("s[1] = %q, want world", *v) + } +} + +// TestBuildTableFieldNativeTypeCoercions pins the client-type mapping: INT32 +// widens to int64, FLOAT widens to float64, an INT64 above 2^53 stays exact +// (native int64, no float roundtrip), DATE renders as yyyy-MM-dd and BLOB as +// 0x-prefixed hex — the same rendering the REST transport produced. +func TestBuildTableFieldNativeTypeCoercions(t *testing.T) { + if v := buildTableField("i", "INT32", []interface{}{int32(7)}).At(0).(*int64); *v != 7 { + t.Fatalf("INT32 = %d, want 7", *v) + } + if v := buildTableField("f", "FLOAT", []interface{}{float32(1.5)}).At(0).(*float64); *v != 1.5 { + t.Fatalf("FLOAT = %v, want 1.5", *v) + } + if v := buildTableField("big", "INT64", []interface{}{int64(9007199254740993)}).At(0).(*int64); *v != 9007199254740993 { + t.Fatalf("INT64 precision lost: %d", *v) + } + date := time.Date(2025, 7, 14, 0, 0, 0, 0, time.UTC) + if v := buildTableField("e", "DATE", []interface{}{date}).At(0).(*string); *v != "2025-07-14" { + t.Fatalf("DATE = %q, want 2025-07-14", *v) + } + if v := buildTableField("d", "BLOB", []interface{}{[]byte{0xca, 0xfe, 0xba, 0xbe}}).At(0).(*string); *v != "0xcafebabe" { + t.Fatalf("BLOB = %q, want 0xcafebabe", *v) + } +} + +func TestBuildTableFieldNullsBecomeNilPointers(t *testing.T) { + f := buildTableField("i", "INT64", []interface{}{int64(1), nil, int64(3)}) + if f.Len() != 3 { + t.Fatalf("expected 3 values, got %d", f.Len()) + } + if v, ok := f.At(1).(*int64); !ok || v != nil { + t.Fatalf("null value should be a nil *int64, got %#v", f.At(1)) + } + if v, ok := f.At(0).(*int64); !ok || v == nil || *v != 1 { + t.Fatalf("first value = %#v, want *int64(1)", f.At(0)) + } +} + +func TestBuildTableFieldUnknownTypeRendersString(t *testing.T) { + f := buildTableField("x", "SOMETHING_NEW", []interface{}{int64(7), "raw", nil}) + if f.Len() != 3 { + t.Fatalf("expected 3 values, got %d", f.Len()) + } + if v, ok := f.At(0).(*string); !ok || v == nil || *v != "7" { + t.Fatalf("numeric coerced to string = %#v, want *string(7)", f.At(0)) + } + if v, ok := f.At(1).(*string); !ok || v == nil || *v != "raw" { + t.Fatalf("string value = %#v, want *string(raw)", f.At(1)) + } + if v, ok := f.At(2).(*string); !ok || v != nil { + t.Fatalf("null value should be nil *string, got %#v", f.At(2)) + } +} + +// TestBuildTableFrameRaggedRowIsSafe checks that a row shorter than the column +// header does not panic and yields equal-length, null-padded fields. +func TestBuildTableFrameRaggedRowIsSafe(t *testing.T) { + ds := &tableQueryDataSet{ + ColumnNames: []string{"time", "s0"}, + DataTypes: []string{"TIMESTAMP", "TEXT"}, + Values: [][]interface{}{ + {ts(1), "a"}, + {ts(2)}, // ragged: missing s0 + }, + } + frame := buildTableFrame(ds) + if len(frame.Fields) != 2 { + t.Fatalf("expected 2 fields, got %d", len(frame.Fields)) + } + if frame.Fields[0].Len() != 2 || frame.Fields[1].Len() != 2 { + t.Fatalf("fields must be equal length (2), got %d and %d", frame.Fields[0].Len(), frame.Fields[1].Len()) + } + if v, ok := frame.Fields[1].At(1).(*string); !ok || v != nil { + t.Fatalf("missing ragged cell should be nil *string, got %#v", frame.Fields[1].At(1)) + } +} + +// TestBuildTableResponseFrameLongToWide pins the multi-device path: a long +// result (time + tag + value), even arriving unsorted, must come back as one +// labeled series per tag value so a time-series panel draws separate lines. +func TestBuildTableResponseFrameLongToWide(t *testing.T) { + ds := &tableQueryDataSet{ + ColumnNames: []string{"time", "device", "temperature"}, + DataTypes: []string{"TIMESTAMP", "STRING", "DOUBLE"}, + Values: [][]interface{}{ + {ts(2000), "d2", float64(22.5)}, + {ts(1000), "d1", float64(11.0)}, + {ts(1000), "d2", float64(21.0)}, + {ts(2000), "d1", float64(12.0)}, + }, + } + + frame := buildTableResponseFrame(ds, "") + if len(frame.Fields) != 3 { + t.Fatalf("expected time + one series per device (3 fields), got %d", len(frame.Fields)) + } + if frame.Fields[0].Len() != 2 { + t.Fatalf("expected 2 wide rows, got %d", frame.Fields[0].Len()) + } + byDevice := map[string][]float64{} + for _, f := range frame.Fields[1:] { + dev := f.Labels["device"] + if dev == "" { + t.Fatalf("value field %q has no device label: %v", f.Name, f.Labels) + } + var vals []float64 + for i := 0; i < f.Len(); i++ { + v, ok := f.At(i).(*float64) + if !ok || v == nil { + t.Fatalf("field %q row %d = %#v, want *float64", f.Name, i, f.At(i)) + } + vals = append(vals, *v) + } + byDevice[dev] = vals + } + if v := byDevice["d1"]; len(v) != 2 || v[0] != 11.0 || v[1] != 12.0 { + t.Fatalf("d1 series = %v, want [11 12]", v) + } + if v := byDevice["d2"]; len(v) != 2 || v[0] != 21.0 || v[1] != 22.5 { + t.Fatalf("d2 series = %v, want [21 22.5]", v) + } +} + +// TestBuildTableResponseFrameTableFormatPreservesOrder pins that the Table +// format neither re-sorts rows (the user's ORDER BY wins) nor pivots tags +// into labels. +func TestBuildTableResponseFrameTableFormatPreservesOrder(t *testing.T) { + ds := &tableQueryDataSet{ + ColumnNames: []string{"time", "device", "temperature"}, + DataTypes: []string{"TIMESTAMP", "STRING", "DOUBLE"}, + Values: [][]interface{}{ + {ts(2000), "d2", float64(22.5)}, + {ts(1000), "d1", float64(11.0)}, + }, + } + + frame := buildTableResponseFrame(ds, tableFormatTable) + if len(frame.Fields) != 3 { + t.Fatalf("expected 3 plain fields, got %d", len(frame.Fields)) + } + first, ok := frame.Fields[0].At(0).(*time.Time) + if !ok || first == nil || !first.Equal(ts(2000)) { + t.Fatalf("row order changed: first time = %#v, want t=2000ms", frame.Fields[0].At(0)) + } + if _, ok := frame.Fields[1].At(0).(*string); !ok { + t.Fatalf("tag column should stay a plain string field in Table format") + } +} + +// TestBuildTableResponseFrameNullTimeFallsBack pins the safety net: when the +// long-to-wide pivot cannot apply (here: a null timestamp), the plain frame is +// returned instead of an error or a panic. +func TestBuildTableResponseFrameNullTimeFallsBack(t *testing.T) { + ds := &tableQueryDataSet{ + ColumnNames: []string{"time", "device", "temperature"}, + DataTypes: []string{"TIMESTAMP", "STRING", "DOUBLE"}, + Values: [][]interface{}{ + {ts(1000), "d1", float64(11.0)}, + {nil, "d2", float64(21.0)}, + }, + } + + frame := buildTableResponseFrame(ds, tableFormatTimeSeries) + if len(frame.Fields) != 3 { + t.Fatalf("expected plain 3-field fallback frame, got %d fields", len(frame.Fields)) + } + if frame.TimeSeriesSchema().Type != data.TimeSeriesTypeLong { + t.Fatalf("fallback frame should still be the long-shaped original") + } +} + +// TestVerifyQueryTableModel pins the table-mode input validation: SQL and +// DATABASE are both required (the latter so every query USEs its database on +// the pooled session and leftover USE state can never leak between panels). +func TestVerifyQueryTableModel(t *testing.T) { + build := func(body string) backend.DataQuery { + return backend.DataQuery{JSON: []byte(body)} + } + + if _, msg := verifyQuery(build(`{"sqlType":"SQL: Table Model","sql":"SELECT 1","database":"db1"}`)); msg != "" { + t.Fatalf("valid table query rejected: %q", msg) + } + if _, msg := verifyQuery(build(`{"sqlType":"SQL: Table Model","sql":" ","database":"db1"}`)); msg != "Input error, SQL is required" { + t.Fatalf("blank sql not rejected: %q", msg) + } + if _, msg := verifyQuery(build(`{"sqlType":"SQL: Table Model","sql":"SELECT 1","database":" "}`)); msg != "Input error, DATABASE is required" { + t.Fatalf("blank database not rejected: %q", msg) + } + if _, msg := verifyQuery(build(`{"sqlType":"SQL: Table Model","sql":"SELECT 1"}`)); msg != "Input error, DATABASE is required" { + t.Fatalf("missing database not rejected: %q", msg) + } +} + +func TestTableRPCEndpoint(t *testing.T) { + cases := []struct { + name string + rpcAddress string + url string + wantHost string + wantPort string + }{ + {name: "derived from http url", url: "http://192.168.1.10:18080", wantHost: "192.168.1.10", wantPort: "6667"}, + {name: "derived from url with trailing slash", url: "http://iotdb.example.com:18080/", wantHost: "iotdb.example.com", wantPort: "6667"}, + {name: "derived from bare host and rest port", url: "192.168.1.10:18080", wantHost: "192.168.1.10", wantPort: "6667"}, + {name: "explicit host and port", rpcAddress: "10.0.0.5:7777", url: "http://x:18080", wantHost: "10.0.0.5", wantPort: "7777"}, + {name: "explicit bare host gets default port", rpcAddress: "10.0.0.5", url: "http://x:18080", wantHost: "10.0.0.5", wantPort: "6667"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + d := &IoTDBDataSource{Ulr: c.url, RPCAddress: c.rpcAddress} + host, port, err := d.tableRPCEndpoint() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if host != c.wantHost || port != c.wantPort { + t.Fatalf("endpoint = %s:%s, want %s:%s", host, port, c.wantHost, c.wantPort) + } + }) + } +} diff --git a/connectors/grafana-plugin/src/ConfigEditor.tsx b/connectors/grafana-plugin/src/ConfigEditor.tsx index fed474dd..9d3b134c 100644 --- a/connectors/grafana-plugin/src/ConfigEditor.tsx +++ b/connectors/grafana-plugin/src/ConfigEditor.tsx @@ -42,6 +42,15 @@ export class ConfigEditor extends PureComponent { onOptionsChange({ ...options, jsonData }); }; + onRPCAddressChange = (event: ChangeEvent) => { + const { onOptionsChange, options } = this.props; + const jsonData = { + ...options.jsonData, + rpcAddress: event.target.value, + }; + onOptionsChange({ ...options, jsonData }); + }; + // Secure field (only sent to the backend) onPasswordChange = (event: ChangeEvent) => { const { onOptionsChange, options } = this.props; @@ -101,6 +110,18 @@ export class ConfigEditor extends PureComponent { onChange={this.onPasswordChange} /> + + + ); diff --git a/connectors/grafana-plugin/src/QueryEditor.tsx b/connectors/grafana-plugin/src/QueryEditor.tsx index 56131bc1..9704c461 100644 --- a/connectors/grafana-plugin/src/QueryEditor.tsx +++ b/connectors/grafana-plugin/src/QueryEditor.tsx @@ -26,7 +26,7 @@ import { FromValue } from './componments/FromValue'; import { WhereValue } from './componments/WhereValue'; import { ControlValue } from './componments/ControlValue'; import { FillValue } from './componments/FillValue'; -import { Segment } from '@grafana/ui'; +import { Input, Segment, TextArea } from '@grafana/ui'; import { toOption } from './functions'; import { GroupByLabel } from './componments/GroupBy'; @@ -46,6 +46,9 @@ interface State { isDropDownList: boolean; sqlType: string; shouldAdd: boolean; + database: string; + sql: string; + format: string; } const selectElement = [ @@ -64,7 +67,8 @@ const selectElement = [ const paths = ['']; const expressions = ['']; -const selectType = ['SQL: Full Customized', 'SQL: Drop-down List']; +const selectType = ['SQL: Full Customized', 'SQL: Drop-down List', 'SQL: Table Model']; +const tableFormats = ['Time series', 'Table']; const commonOption: SelectableValue = { label: '*', value: '*' }; const commonOptionDou: SelectableValue = { label: '**', value: '**' }; type Props = QueryEditorProps; @@ -87,6 +91,9 @@ export class QueryEditor extends PureComponent { isDropDownList: false, sqlType: selectType[0], shouldAdd: true, + database: '', + sql: '', + format: tableFormats[0], }; @@ -134,7 +141,26 @@ export class QueryEditor extends PureComponent { onChange({ ...query, groupBy: g }); }; - + onDatabaseChange = (event: ChangeEvent) => { + const { onChange, query } = this.props; + const database = event.target.value; + this.setState({ database }); + onChange({ ...query, database }); + }; + + onSqlChange = (event: ChangeEvent) => { + const { onChange, query } = this.props; + const sql = event.target.value; + this.setState({ sql }); + onChange({ ...query, sql }); + }; + + onFormatChange = ({ value: value = tableFormats[0] }: SelectableValue) => { + const { onChange, query } = this.props; + this.setState({ format: value }); + onChange({ ...query, format: value }); + }; + onSelectTypeChange = (event: ChangeEvent) => { const { onChange, query } = this.props; onChange({ ...query }); @@ -175,7 +201,13 @@ export class QueryEditor extends PureComponent { componentDidMount() { if (this.props.query.sqlType) { - this.setState({ isDropDownList: this.props.query.isDropDownList, sqlType: this.props.query.sqlType }); + this.setState({ + isDropDownList: this.props.query.isDropDownList, + sqlType: this.props.query.sqlType, + database: this.props.query.database ?? '', + sql: this.props.query.sql ?? '', + format: this.props.query.format ?? tableFormats[0], + }); } else { this.props.query.sqlType = selectType[0]; } @@ -231,7 +263,7 @@ export class QueryEditor extends PureComponent { condition: '', }); onChange({ ...query, sqlType: value, isDropDownList: false }); - } else { + } else if (value === selectType[1]) { this.props.query.sqlType = selectType[1]; this.props.query.expression = ['']; this.props.query.prefixPath = ['']; @@ -247,6 +279,22 @@ export class QueryEditor extends PureComponent { control: '', }); onChange({ ...query, sqlType: value, isDropDownList: true }); + } else { + this.props.query.sqlType = selectType[2]; + this.props.query.expression = ['']; + this.props.query.prefixPath = ['']; + this.props.query.condition = ''; + this.props.query.control = ''; + this.props.query.isDropDownList = false; + this.setState({ + isDropDownList: false, + sqlType: selectType[2], + expression: [''], + prefixPath: [''], + condition: '', + control: '', + }); + onChange({ ...query, sqlType: value, isDropDownList: false }); } }} options={selectType.map(toOption)} @@ -254,7 +302,7 @@ export class QueryEditor extends PureComponent { className="query-keyword width-10" /> - {!this.state.isDropDownList && ( + {this.state.sqlType === selectType[0] && ( <>
@@ -335,6 +383,39 @@ export class QueryEditor extends PureComponent {
)} + {this.state.sqlType === selectType[2] && ( + <> +
+ + + +
+
+ +