This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
The following commit(s) were added to refs/heads/master by this push:
new d11e1c0 support dynamic IoTDB table-model variables (#126)
d11e1c0 is described below
commit d11e1c03d4a28dfcd90732bc3a47a0b3cbd38fe9
Author: Zhao Xinqi <[email protected]>
AuthorDate: Fri Aug 21 17:43:09 2026 +0800
support dynamic IoTDB table-model variables (#126)
---
.../pkg/plugin/iotdb_resource_handler.go | 122 +++++-
connectors/grafana-plugin/pkg/plugin/plugin.go | 23 +-
.../grafana-plugin/pkg/plugin/table_query.go | 88 ++++-
.../pkg/plugin/table_variable_test.go | 411 +++++++++++++++++++++
4 files changed, 623 insertions(+), 21 deletions(-)
diff --git a/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go
b/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go
index f0de10f..9154ac3 100644
--- a/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go
+++ b/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go
@@ -20,20 +20,30 @@ package plugin
import (
"bytes"
"encoding/json"
+ "errors"
"io"
"io/ioutil"
"net/http"
+ "strings"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
"github.com/grafana/grafana-plugin-sdk-go/backend/resource/httpadapter"
)
-func iotdbResourceHandler(authorization string, httpClient *http.Client)
backend.CallResourceHandler {
+// tableVariablePrefix marks a template-variable query as a table-model query.
+// A variable query of the form "table:<database>:<SQL>" is executed through
the
+// IoTDB table-model RPC client instead of the legacy tree-model REST endpoint.
+const tableVariablePrefix = "table:"
+
+// iotdbResourceHandler wires the plugin resource endpoints. It is a method so
+// the table-model variable path can reach the datasource's native-client
+// session pool.
+func (d *IoTDBDataSource) iotdbResourceHandler(authorization string,
httpClient *http.Client) backend.CallResourceHandler {
mux := http.NewServeMux()
- mux.Handle("/getVariables", getVariables(authorization, httpClient))
- mux.Handle("/getNodes", getNodes(authorization, httpClient))
+ mux.Handle("/getVariables", d.getVariables(authorization, httpClient))
+ mux.Handle("/getNodes", d.getNodes(authorization, httpClient))
return httpadapter.New(mux)
}
@@ -43,7 +53,6 @@ type queryReq struct {
}
type nodeReq struct {
Data []string `json:"data"`
- Url string `json:"url"`
}
type queryResp struct {
@@ -51,22 +60,41 @@ type queryResp struct {
Message string `json:"message"`
}
-func getVariables(authorization string, httpClient *http.Client) http.Handler {
+func (d *IoTDBDataSource) getVariables(authorization string, httpClient
*http.Client) http.Handler {
fn := func(w http.ResponseWriter, r *http.Request) {
- var url = r.FormValue("url")
- var sql = r.FormValue("sql")
if r.Method != http.MethodGet {
http.NotFound(w, r)
return
}
+ var sql = r.FormValue("sql")
+
+ // table:<database>:<SQL> variables run through the table-model
RPC
+ // client; every other query keeps the legacy tree-model
behavior.
+ if strings.HasPrefix(strings.TrimSpace(sql),
tableVariablePrefix) {
+ d.handleTableVariableQuery(w, r, sql)
+ return
+ }
+
var queryReq = &queryReq{Sql: sql}
qpJson, _ := json.Marshal(queryReq)
reader := bytes.NewReader(qpJson)
client := &http.Client{}
- request, _ := http.NewRequest(http.MethodPost,
url+"/grafana/v1/variable", reader)
+ // The tree-model endpoint is always the datasource's
configured URL
+ // (d.Ulr), never the client-supplied "url" query parameter, so
a caller
+ // cannot redirect this request to an arbitrary host (SSRF).
+ request, err := http.NewRequest(http.MethodPost,
DataSourceUrlHandler(d.Ulr)+"/grafana/v1/variable", reader)
+ if err != nil {
+ writeJSONError(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
request.Header.Set("Content-Type", "application/json")
request.Header.Add("Authorization", authorization)
- rsp, _ := client.Do(request)
+ rsp, err := client.Do(request)
+ if err != nil {
+ log.DefaultLogger.Error("Data source is not working
properly", err)
+ writeJSONError(w, http.StatusInternalServerError,
err.Error())
+ return
+ }
body, err := io.ReadAll(rsp.Body)
if err != nil {
log.DefaultLogger.Error("Data source is not working
properly", err)
@@ -108,7 +136,65 @@ func getVariables(authorization string, httpClient
*http.Client) http.Handler {
return http.HandlerFunc(fn)
}
-func getNodes(authorization string, client *http.Client) http.Handler {
+// handleTableVariableQuery runs a table-model variable query and writes the
+// single-column result as a JSON string array (the shape Grafana's variable
+// dropdown expects), or a JSON error when parsing or execution fails.
+func (d *IoTDBDataSource) handleTableVariableQuery(w http.ResponseWriter, r
*http.Request, query string) {
+ database, sql, err := parseTableVariableQuery(query)
+ if err != nil {
+ writeJSONError(w, http.StatusBadRequest, err.Error())
+ return
+ }
+ sql = d.expandVariableMacros(sql)
+ values, err := d.tableVariableValues(r.Context(), database, sql)
+ if err != nil {
+ log.DefaultLogger.Error("table-model variable query failed",
"err", err)
+ writeJSONError(w, http.StatusInternalServerError, err.Error())
+ return
+ }
+ writeJSON(w, values)
+}
+
+// parseTableVariableQuery splits a table-model variable query of the form
+// "table:<database>:<SQL>" into its database and SQL parts. Both parts are
+// required; an empty or missing part is an error rather than a fallback to the
+// legacy path.
+func parseTableVariableQuery(query string) (database, sql string, err error) {
+ body := strings.TrimSpace(query)
+ if !strings.HasPrefix(body, tableVariablePrefix) {
+ return "", "", errors.New("table-model variable query must
start with table:")
+ }
+ rest := body[len(tableVariablePrefix):]
+ separator := strings.IndexByte(rest, ':')
+ if separator < 0 {
+ return "", "", errors.New("table-model variable query is
missing the database:SQL separator")
+ }
+ database = strings.TrimSpace(rest[:separator])
+ sql = strings.TrimSpace(rest[separator+1:])
+ if database == "" {
+ return "", "", errors.New("table-model variable query requires
a database")
+ }
+ if sql == "" {
+ return "", "", errors.New("table-model variable query requires
SQL")
+ }
+ return database, sql, nil
+}
+
+// writeJSON writes a value as a JSON response with the default 200 status.
+func writeJSON(w http.ResponseWriter, value interface{}) {
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(value)
+}
+
+// writeJSONError writes a machine-readable JSON error with an explicit HTTP
+// status so a failed variable query is never mistaken for an empty result.
+func writeJSONError(w http.ResponseWriter, status int, message string) {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(status)
+ _ = json.NewEncoder(w).Encode(queryResp{Code: status, Message: message})
+}
+
+func (d *IoTDBDataSource) getNodes(authorization string, client *http.Client)
http.Handler {
fn := func(w http.ResponseWriter, r *http.Request) {
s, _ := ioutil.ReadAll(r.Body)
if r.Method != http.MethodPost {
@@ -124,10 +210,22 @@ func getNodes(authorization string, client *http.Client)
http.Handler {
qpJson, _ := json.Marshal(nodeReq.Data)
reader := bytes.NewReader(qpJson)
- request, _ := http.NewRequest(http.MethodPost,
nodeReq.Url+"/grafana/v1/node", reader)
+ // The node endpoint is always the datasource's configured URL
(d.Ulr),
+ // never a client-supplied URL, so a caller cannot redirect
this request
+ // to an arbitrary host (SSRF).
+ request, err := http.NewRequest(http.MethodPost,
DataSourceUrlHandler(d.Ulr)+"/grafana/v1/node", reader)
+ if err != nil {
+ http.Error(w, err.Error(),
http.StatusInternalServerError)
+ return
+ }
request.Header.Set("Content-Type", "application/json")
request.Header.Add("Authorization", authorization)
- rsp, _ := client.Do(request)
+ rsp, err := client.Do(request)
+ if err != nil {
+ log.DefaultLogger.Error("Data source is not working
properly", err)
+ http.Error(w, err.Error(),
http.StatusInternalServerError)
+ return
+ }
body, err := io.ReadAll(rsp.Body)
if err != nil {
log.DefaultLogger.Error("Data source is not working
properly", err)
diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go
b/connectors/grafana-plugin/pkg/plugin/plugin.go
index 8e527c4..e31e048 100644
--- a/connectors/grafana-plugin/pkg/plugin/plugin.go
+++ b/connectors/grafana-plugin/pkg/plugin/plugin.go
@@ -73,7 +73,17 @@ func ApacheIoTDBDatasource(ctx context.Context, d
backend.DataSourceInstanceSett
authorization = "Basic " +
base64.StdEncoding.EncodeToString([]byte(dm.Username+":"+password))
}
password := d.DecryptedSecureJSONData["password"]
- return &IoTDBDataSource{CallResourceHandler:
iotdbResourceHandler(authorization, httpClient), Username: dm.Username, Ulr:
dm.Url, RPCAddress: dm.RPCAddress, password: password, httpClient: httpClient},
nil
+ ds := &IoTDBDataSource{
+ Username: dm.Username,
+ Ulr: dm.Url,
+ RPCAddress: dm.RPCAddress,
+ password: password,
+ httpClient: httpClient,
+ }
+ // The resource handler is a method so the table-model variable path can
+ // reach the native-client session pool.
+ ds.CallResourceHandler = ds.iotdbResourceHandler(authorization,
httpClient)
+ return ds, nil
}
// SampleDatasource is an example datasource which can respond to data
queries, reports
@@ -90,9 +100,20 @@ type IoTDBDataSource struct {
// getTablePool on the first table query.
tablePoolMu sync.Mutex
tablePool *client.TableSessionPool
+
// tableQueryRunner is replaceable in tests so queryTableModel's
response
// behavior can be exercised without a live IoTDB RPC service.
tableQueryRunner func(context.Context, *queryParam)
(*tableQueryDataSet, error)
+
+ // tableExecutor runs a table-model statement and returns the fetched
+ // dataset. A nil value selects the real RPC executor; tests substitute
a
+ // fake to exercise the variable-query path without a live server.
+ tableExecutor func(ctx context.Context, database, sql string)
(*tableQueryDataSet, error)
+
+ // now returns the current instant used to compute the node-liveness
window
+ // for table-model template-variable queries. A nil value selects
time.Now;
+ // tests substitute a fixed clock so the window is deterministic.
+ now func() time.Time
}
// Dispose here tells plugin SDK that plugin wants to clean up resources when
a new instance
diff --git a/connectors/grafana-plugin/pkg/plugin/table_query.go
b/connectors/grafana-plugin/pkg/plugin/table_query.go
index 1a5c955..b03e3cc 100644
--- a/connectors/grafana-plugin/pkg/plugin/table_query.go
+++ b/connectors/grafana-plugin/pkg/plugin/table_query.go
@@ -80,8 +80,20 @@ var (
intervalMSRe = regexp.MustCompile(`\$__interval_ms`)
)
+// activeFromRe matches $__activeFrom, the lower bound of the node-liveness
+// window used by table-model template-variable queries. It is a variable-path
+// macro (not a panel-query macro) and is expanded by expandVariableMacros.
+var activeFromRe = regexp.MustCompile(`\$__activeFrom\b`)
+
const invalidIntervalMacroMessage = "Grafana query interval must be positive
when $__interval or $__interval_ms is used"
+// nodeActiveTTL is the template-variable liveness window. A node is considered
+// active only if its most recent sample falls within the last nodeActiveTTL.
+// The bridge writes samples with their Prometheus scrape timestamp, so this is
+// "the node produced a scrape within the last nodeActiveTTL"; a node that
stops
+// producing samples ages out of the window and disappears from the variable.
+const nodeActiveTTL = 5 * time.Minute
+
// 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
@@ -91,6 +103,26 @@ func formatTimeLiteral(ms int64) string {
return time.UnixMilli(ms).UTC().Format("2006-01-02T15:04:05.000") +
"+00:00"
}
+// currentTime returns the injectable clock, or the real wall clock when no
+// clock is configured.
+func (d *IoTDBDataSource) currentTime() time.Time {
+ if d.now != nil {
+ return d.now()
+ }
+ return time.Now()
+}
+
+// expandVariableMacros rewrites the template-variable-only macros before a
+// table-model variable query runs. $__activeFrom becomes the "now - TTL"
+// instant rendered as an ISO 8601 UTC timestamp literal — the same literal
form
+// panel queries receive for $__timeFrom — so a variable query can restrict its
+// result to recently active nodes without the server evaluating a time
+// function of its own.
+func (d *IoTDBDataSource) expandVariableMacros(sql string) string {
+ from := d.currentTime().Add(-nodeActiveTTL).UnixMilli()
+ return activeFromRe.ReplaceAllString(sql, formatTimeLiteral(from))
+}
+
// expandTableMacros rewrites the Grafana time and interval macros a dashboard
// author can put in table-model SQL.
//
@@ -335,6 +367,23 @@ func (d *IoTDBDataSource) queryTableModel(ctx
context.Context, qp *queryParam) b
// executeTableQuery runs and fetches one table-model query. queryTableModel
// owns response semantics so the same zero-row path is covered in tests.
func (d *IoTDBDataSource) executeTableQuery(ctx context.Context, qp
*queryParam) (*tableQueryDataSet, error) {
+ sql, err := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime,
qp.IntervalMS)
+ if err != nil {
+ return nil, err
+ }
+ return d.executeTableStatement(ctx, qp.Database, sql)
+}
+
+// executeTableStatement runs a table-model SQL statement against the given
+// database on a pooled native-client session and returns the fetched dataset.
+// The database is USEd on the session so a statement's table references
resolve
+// against it regardless of any session state left by earlier queries. When
+// tableExecutor is non-nil it is used instead of the real RPC path.
+func (d *IoTDBDataSource) executeTableStatement(ctx context.Context, database,
sql string) (*tableQueryDataSet, error) {
+ if d.tableExecutor != nil {
+ return d.tableExecutor(ctx, database, sql)
+ }
+
pool, err := d.getTablePool()
if err != nil {
return nil, err
@@ -350,8 +399,8 @@ func (d *IoTDBDataSource) executeTableQuery(ctx
context.Context, qp *queryParam)
}
}()
- if database := strings.TrimSpace(qp.Database); database != "" {
- if err := session.ExecuteNonQueryStatement("USE " +
quoteTableIdentifier(database)); err != nil {
+ if db := strings.TrimSpace(database); db != "" {
+ if err := session.ExecuteNonQueryStatement("USE " +
quoteTableIdentifier(db)); err != nil {
return nil, err
}
}
@@ -362,10 +411,6 @@ func (d *IoTDBDataSource) executeTableQuery(ctx
context.Context, qp *queryParam)
timeout = ms
}
}
- sql, err := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime,
qp.IntervalMS)
- if err != nil {
- return nil, err
- }
resultSet, err := session.ExecuteQueryStatement(sql, &timeout)
if err != nil {
return nil, err
@@ -376,11 +421,38 @@ func (d *IoTDBDataSource) executeTableQuery(ctx
context.Context, qp *queryParam)
}
}()
- dataSet, err := fetchTableDataSet(resultSet)
+ return fetchTableDataSet(resultSet)
+}
+
+// tableVariableValues runs a table-model template-variable query and returns
+// its single string column as an ordered, NULL-free slice for Grafana's
+// variable dropdown.
+func (d *IoTDBDataSource) tableVariableValues(ctx context.Context, database,
sql string) ([]string, error) {
+ dataSet, err := d.executeTableStatement(ctx, database, sql)
if err != nil {
return nil, err
}
- return dataSet, nil
+ return tableVariableStrings(dataSet)
+}
+
+// tableVariableStrings extracts the values of the single column a template
+// variable query must project. NULL cells are skipped and non-string cells are
+// rendered with the table-mode string coercion, preserving server order.
+func tableVariableStrings(dataSet *tableQueryDataSet) ([]string, error) {
+ if len(dataSet.ColumnNames) != 1 {
+ return nil, fmt.Errorf("template variable query must project
exactly one column, got %d", len(dataSet.ColumnNames))
+ }
+ values := make([]string, 0, len(dataSet.Values))
+ for _, row := range dataSet.Values {
+ if len(row) == 0 {
+ continue
+ }
+ if row[0] == nil {
+ continue
+ }
+ values = append(values, toString(row[0]))
+ }
+ return values, nil
}
// buildTableResponseFrame turns a fetched dataset into the response frame,
diff --git a/connectors/grafana-plugin/pkg/plugin/table_variable_test.go
b/connectors/grafana-plugin/pkg/plugin/table_variable_test.go
new file mode 100644
index 0000000..b444c32
--- /dev/null
+++ b/connectors/grafana-plugin/pkg/plugin/table_variable_test.go
@@ -0,0 +1,411 @@
+/*
+ * 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/base64"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestParseTableVariableQuery(t *testing.T) {
+ database, sql, err :=
parseTableVariableQuery("table:metrics_validation:SELECT DISTINCT instance FROM
sys_cpu_cores")
+ if err != nil {
+ t.Fatalf("valid query rejected: %v", err)
+ }
+ if database != "metrics_validation" {
+ t.Fatalf("database = %q, want metrics_validation", database)
+ }
+ if sql != "SELECT DISTINCT instance FROM sys_cpu_cores" {
+ t.Fatalf("sql = %q", sql)
+ }
+}
+
+func TestParseTableVariableQueryTrimsWhitespace(t *testing.T) {
+ database, sql, err := parseTableVariableQuery(" table:
metrics_validation : SELECT 1 ")
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if database != "metrics_validation" || sql != "SELECT 1" {
+ t.Fatalf("database = %q, sql = %q", database, sql)
+ }
+}
+
+func TestParseTableVariableQueryRejectsNonTableQuery(t *testing.T) {
+ if _, _, err := parseTableVariableQuery("SELECT 1"); err == nil {
+ t.Fatalf("non-table query should not parse as a table query")
+ }
+}
+
+func TestParseTableVariableQueryRejectsEmptyDatabase(t *testing.T) {
+ if _, _, err := parseTableVariableQuery("table::SELECT 1"); err == nil
|| !strings.Contains(err.Error(), "database") {
+ t.Fatalf("empty database error = %v", err)
+ }
+}
+
+func TestParseTableVariableQueryRejectsEmptySQL(t *testing.T) {
+ if _, _, err := parseTableVariableQuery("table:metrics_validation:");
err == nil || !strings.Contains(err.Error(), "SQL") {
+ t.Fatalf("empty SQL error = %v", err)
+ }
+}
+
+func TestParseTableVariableQueryRejectsMissingSeparator(t *testing.T) {
+ if _, _, err := parseTableVariableQuery("table:metrics_validation");
err == nil {
+ t.Fatalf("missing separator should be an error")
+ }
+}
+
+func TestTableVariableStringsSingleColumnPreservesOrder(t *testing.T) {
+ dataSet := &tableQueryDataSet{
+ ColumnNames: []string{"instance"},
+ DataTypes: []string{"STRING"},
+ Values: [][]interface{}{
+ {"node-a.example.test:9091"},
+ {"node-b.example.test:9091"},
+ {"node-c.example.test:9091"},
+ },
+ }
+ got, err := tableVariableStrings(dataSet)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ want := []string{"node-a.example.test:9091",
"node-b.example.test:9091", "node-c.example.test:9091"}
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("values = %#v, want %#v", got, want)
+ }
+}
+
+func TestTableVariableStringsSkipsNulls(t *testing.T) {
+ dataSet := &tableQueryDataSet{
+ ColumnNames: []string{"instance"},
+ DataTypes: []string{"STRING"},
+ Values: [][]interface{}{
+ {"a"},
+ {nil},
+ {"b"},
+ },
+ }
+ got, err := tableVariableStrings(dataSet)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if !reflect.DeepEqual(got, []string{"a", "b"}) {
+ t.Fatalf("values = %#v, want [a b]", got)
+ }
+}
+
+func TestTableVariableStringsCoercesNonStrings(t *testing.T) {
+ dataSet := &tableQueryDataSet{
+ ColumnNames: []string{"node_num"},
+ DataTypes: []string{"INT64"},
+ Values: [][]interface{}{
+ {int64(7)},
+ },
+ }
+ got, err := tableVariableStrings(dataSet)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if !reflect.DeepEqual(got, []string{"7"}) {
+ t.Fatalf("values = %#v, want [7]", got)
+ }
+}
+
+func TestTableVariableStringsRejectsMultiColumn(t *testing.T) {
+ dataSet := &tableQueryDataSet{
+ ColumnNames: []string{"cluster", "instance"},
+ DataTypes: []string{"STRING", "STRING"},
+ Values: [][]interface{}{{"a", "b"}},
+ }
+ if _, err := tableVariableStrings(dataSet); err == nil ||
!strings.Contains(err.Error(), "exactly one column") {
+ t.Fatalf("multi-column error = %v", err)
+ }
+}
+
+func TestTableVariableValuesCallsExecutor(t *testing.T) {
+ var gotDB, gotSQL string
+ d := &IoTDBDataSource{
+ tableExecutor: func(ctx context.Context, database, sql string)
(*tableQueryDataSet, error) {
+ gotDB, gotSQL = database, sql
+ return &tableQueryDataSet{
+ ColumnNames: []string{"instance"},
+ DataTypes: []string{"STRING"},
+ Values:
[][]interface{}{{"node-a.example.test:9091"}},
+ }, nil
+ },
+ }
+ values, err := d.tableVariableValues(context.Background(),
"metrics_validation", "SELECT DISTINCT instance FROM sys_cpu_cores")
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if gotDB != "metrics_validation" {
+ t.Fatalf("executor database = %q, want metrics_validation",
gotDB)
+ }
+ if gotSQL != "SELECT DISTINCT instance FROM sys_cpu_cores" {
+ t.Fatalf("executor sql = %q", gotSQL)
+ }
+ if !reflect.DeepEqual(values, []string{"node-a.example.test:9091"}) {
+ t.Fatalf("values = %#v", values)
+ }
+}
+
+func TestTableVariableValuesPropagatesExecutorError(t *testing.T) {
+ d := &IoTDBDataSource{
+ tableExecutor: func(ctx context.Context, database, sql string)
(*tableQueryDataSet, error) {
+ return nil, errors.New("connection refused")
+ },
+ }
+ if _, err := d.tableVariableValues(context.Background(), "db", "SELECT
1"); err == nil || !strings.Contains(err.Error(), "connection refused") {
+ t.Fatalf("error = %v, want connection refused", err)
+ }
+}
+
+func TestGetVariablesTableModelQuery(t *testing.T) {
+ var gotDB, gotSQL string
+ d := &IoTDBDataSource{
+ tableExecutor: func(ctx context.Context, database, sql string)
(*tableQueryDataSet, error) {
+ gotDB, gotSQL = database, sql
+ return &tableQueryDataSet{
+ ColumnNames: []string{"instance"},
+ DataTypes: []string{"STRING"},
+ Values: [][]interface{}{
+ {"node-a.example.test:9091"},
+ {nil},
+ {"node-b.example.test:9091"},
+ },
+ }, nil
+ },
+ }
+ handler := d.getVariables("", http.DefaultClient)
+ request := httptest.NewRequest(http.MethodGet,
"/getVariables?url="+url.QueryEscape("http://iotdb:18080";)+"&sql="+url.QueryEscape("table:metrics_validation:SELECT
DISTINCT instance FROM sys_cpu_cores"), nil)
+ recorder := httptest.NewRecorder()
+ handler.ServeHTTP(recorder, request)
+
+ if recorder.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200; body = %s", recorder.Code,
recorder.Body.String())
+ }
+ if gotDB != "metrics_validation" || gotSQL != "SELECT DISTINCT instance
FROM sys_cpu_cores" {
+ t.Fatalf("executor got database %q, sql %q", gotDB, gotSQL)
+ }
+ var values []string
+ if err := json.Unmarshal(recorder.Body.Bytes(), &values); err != nil {
+ t.Fatalf("response is not a JSON array: %v; body = %s", err,
recorder.Body.String())
+ }
+ if !reflect.DeepEqual(values, []string{"node-a.example.test:9091",
"node-b.example.test:9091"}) {
+ t.Fatalf("values = %#v", values)
+ }
+}
+
+func TestGetVariablesTableModelMultiColumnError(t *testing.T) {
+ d := &IoTDBDataSource{
+ tableExecutor: func(ctx context.Context, database, sql string)
(*tableQueryDataSet, error) {
+ return &tableQueryDataSet{
+ ColumnNames: []string{"cluster", "instance"},
+ DataTypes: []string{"STRING", "STRING"},
+ Values: [][]interface{}{{"a", "b"}},
+ }, nil
+ },
+ }
+ handler := d.getVariables("", http.DefaultClient)
+ request := httptest.NewRequest(http.MethodGet,
"/getVariables?url=x&sql="+url.QueryEscape("table:db:SELECT cluster, instance
FROM t"), nil)
+ recorder := httptest.NewRecorder()
+ handler.ServeHTTP(recorder, request)
+
+ if recorder.Code != http.StatusInternalServerError {
+ t.Fatalf("status = %d, want 500", recorder.Code)
+ }
+ var body queryResp
+ if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
+ t.Fatalf("error body is not JSON: %v; body = %s", err,
recorder.Body.String())
+ }
+ if body.Message == "" || !strings.Contains(body.Message, "exactly one
column") {
+ t.Fatalf("error message = %q", body.Message)
+ }
+}
+
+func TestGetVariablesTableModelParseError(t *testing.T) {
+ d := &IoTDBDataSource{}
+ handler := d.getVariables("", http.DefaultClient)
+ request := httptest.NewRequest(http.MethodGet,
"/getVariables?url=x&sql="+url.QueryEscape("table:metrics_validation:"), nil)
+ recorder := httptest.NewRecorder()
+ handler.ServeHTTP(recorder, request)
+
+ if recorder.Code != http.StatusBadRequest {
+ t.Fatalf("status = %d, want 400", recorder.Code)
+ }
+}
+
+func TestGetVariablesLegacyPath(t *testing.T) {
+ var gotPath, gotBody string
+ legacy := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ gotPath = r.URL.Path
+ var incoming queryReq
+ _ = json.NewDecoder(r.Body).Decode(&incoming)
+ gotBody = incoming.Sql
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`["a","b"]`))
+ }))
+ defer legacy.Close()
+
+ // The request goes to the datasource's configured URL (d.Ulr); the
+ // client-supplied "url" query parameter is ignored so a caller cannot
+ // redirect the request (SSRF).
+ d := &IoTDBDataSource{Ulr: legacy.URL}
+ handler := d.getVariables("Bearer test", http.DefaultClient)
+ request := httptest.NewRequest(http.MethodGet,
"/getVariables?url="+url.QueryEscape("http://169.254.169.254";)+"&sql="+url.QueryEscape("show
timeseries"), nil)
+ recorder := httptest.NewRecorder()
+ handler.ServeHTTP(recorder, request)
+
+ if recorder.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200", recorder.Code)
+ }
+ if gotPath != "/grafana/v1/variable" {
+ t.Fatalf("legacy path = %q, want /grafana/v1/variable", gotPath)
+ }
+ if gotBody != "show timeseries" {
+ t.Fatalf("legacy body = %q, want show timeseries", gotBody)
+ }
+ var values []string
+ if err := json.Unmarshal(recorder.Body.Bytes(), &values); err != nil ||
!reflect.DeepEqual(values, []string{"a", "b"}) {
+ t.Fatalf("legacy response = %s (err %v)",
recorder.Body.String(), err)
+ }
+}
+
+func TestTableVariableQueryDoesNotLeakCredentials(t *testing.T) {
+ const secret = "sup3r-s3cret-password"
+ authorization := "Basic " +
base64.StdEncoding.EncodeToString([]byte("root:"+secret))
+ d := &IoTDBDataSource{
+ password: secret,
+ tableExecutor: func(ctx context.Context, database, sql string)
(*tableQueryDataSet, error) {
+ return &tableQueryDataSet{
+ ColumnNames: []string{"instance"},
+ DataTypes: []string{"STRING"},
+ Values:
[][]interface{}{{"node-a.example.test:9091"}},
+ }, nil
+ },
+ }
+ handler := d.getVariables(authorization, http.DefaultClient)
+ request := httptest.NewRequest(http.MethodGet,
"/getVariables?url=x&sql="+url.QueryEscape("table:db:SELECT DISTINCT instance
FROM t"), nil)
+ recorder := httptest.NewRecorder()
+ handler.ServeHTTP(recorder, request)
+
+ body := recorder.Body.String()
+ if strings.Contains(body, secret) {
+ t.Fatalf("response leaked the password: %s", body)
+ }
+ if strings.Contains(body, authorization) {
+ t.Fatalf("response leaked the authorization header: %s", body)
+ }
+ var values []string
+ if err := json.Unmarshal(recorder.Body.Bytes(), &values); err != nil ||
!reflect.DeepEqual(values, []string{"node-a.example.test:9091"}) {
+ t.Fatalf("response = %s (err %v)", body, err)
+ }
+}
+
+func TestExpandVariableMacrosActiveFrom(t *testing.T) {
+ fixed := time.Date(2026, 8, 20, 12, 0, 0, 0, time.UTC)
+ d := &IoTDBDataSource{now: func() time.Time { return fixed }}
+ sql := "SELECT DISTINCT instance FROM sys_cpu_cores WHERE time >=
$__activeFrom AND instance <> ''"
+ got := d.expandVariableMacros(sql)
+ if strings.Contains(got, "$__activeFrom") {
+ t.Fatalf("macro left unexpanded: %s", got)
+ }
+ want := formatTimeLiteral(fixed.Add(-nodeActiveTTL).UnixMilli())
+ if !strings.Contains(got, "time >= "+want) {
+ t.Fatalf("expanded SQL = %q, want lower bound %q", got, want)
+ }
+}
+
+func TestExpandVariableMacrosLeavesOtherSQLAlone(t *testing.T) {
+ d := &IoTDBDataSource{now: func() time.Time { return time.UnixMilli(0)
}}
+ sql := "SELECT DISTINCT cluster FROM sys_cpu_cores WHERE cluster <> ''"
+ if got := d.expandVariableMacros(sql); got != sql {
+ t.Fatalf("SQL without $__activeFrom changed: %q", got)
+ }
+}
+
+func TestHandleTableVariableQueryExpandsActiveFrom(t *testing.T) {
+ fixed := time.Date(2026, 8, 20, 12, 0, 0, 0, time.UTC)
+ var gotSQL string
+ d := &IoTDBDataSource{
+ now: func() time.Time { return fixed },
+ tableExecutor: func(ctx context.Context, database, sql string)
(*tableQueryDataSet, error) {
+ gotSQL = sql
+ return &tableQueryDataSet{
+ ColumnNames: []string{"instance"},
+ DataTypes: []string{"STRING"},
+ Values:
[][]interface{}{{"node-a.example.test:9091"}},
+ }, nil
+ },
+ }
+ handler := d.getVariables("", http.DefaultClient)
+ request := httptest.NewRequest(http.MethodGet,
"/getVariables?url=x&sql="+url.QueryEscape("table:db:SELECT DISTINCT instance
FROM t WHERE time >= $__activeFrom"), nil)
+ recorder := httptest.NewRecorder()
+ handler.ServeHTTP(recorder, request)
+
+ if recorder.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200; body = %s", recorder.Code,
recorder.Body.String())
+ }
+ if strings.Contains(gotSQL, "$__activeFrom") {
+ t.Fatalf("executor received unexpanded macro: %q", gotSQL)
+ }
+ want := formatTimeLiteral(fixed.Add(-nodeActiveTTL).UnixMilli())
+ if !strings.Contains(gotSQL, "time >= "+want) {
+ t.Fatalf("executor SQL = %q, want lower bound %q", gotSQL, want)
+ }
+}
+
+func TestGetNodesUsesConfiguredURL(t *testing.T) {
+ var gotPath, gotBody string
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ gotPath = r.URL.Path
+ body, _ := io.ReadAll(r.Body)
+ gotBody = string(body)
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`["root.sg"]`))
+ }))
+ defer server.Close()
+
+ // The node endpoint goes to the datasource's configured URL (d.Ulr);
the
+ // client-supplied "url" field is ignored so a caller cannot redirect
the
+ // request (SSRF).
+ d := &IoTDBDataSource{Ulr: server.URL}
+ handler := d.getNodes("", http.DefaultClient)
+ request := httptest.NewRequest(http.MethodPost, "/getNodes",
strings.NewReader(`{"data":["root.sg"],"url":"http://169.254.169.254"}`))
+ recorder := httptest.NewRecorder()
+ handler.ServeHTTP(recorder, request)
+
+ if recorder.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200; body = %s", recorder.Code,
recorder.Body.String())
+ }
+ if gotPath != "/grafana/v1/node" {
+ t.Fatalf("node path = %q, want /grafana/v1/node", gotPath)
+ }
+ if gotBody != `["root.sg"]` {
+ t.Fatalf("node body = %q, want [\"root.sg\"]", gotBody)
+ }
+}
