From 15cf5a10d1a0bad57ffdb863893cf567c74b822f Mon Sep 17 00:00:00 2001 From: zxq <3322351820@qq.com> Date: Fri, 21 Aug 2026 11:18:31 +0800 Subject: [PATCH 1/2] support dynamic IoTDB table-model variables --- .../pkg/plugin/iotdb_resource_handler.go | 86 +++- .../grafana-plugin/pkg/plugin/plugin.go | 22 +- .../grafana-plugin/pkg/plugin/table_query.go | 111 +++++- .../pkg/plugin/table_variable_test.go | 376 ++++++++++++++++++ 4 files changed, 570 insertions(+), 25 deletions(-) create mode 100644 connectors/grafana-plugin/pkg/plugin/table_variable_test.go diff --git a/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go b/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go index f0de10f7..dcd4511b 100644 --- a/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go +++ b/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go @@ -20,19 +20,29 @@ 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::" 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("/getVariables", d.getVariables(authorization, httpClient)) mux.Handle("/getNodes", getNodes(authorization, httpClient)) return httpadapter.New(mux) @@ -51,14 +61,22 @@ 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 url = r.FormValue("url") + var sql = r.FormValue("sql") + + // table:: 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) @@ -108,6 +126,64 @@ func getVariables(authorization string, httpClient *http.Client) http.Handler { return http.HandlerFunc(fn) } +// 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::" 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 getNodes(authorization string, client *http.Client) http.Handler { fn := func(w http.ResponseWriter, r *http.Request) { s, _ := ioutil.ReadAll(r.Body) diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go b/connectors/grafana-plugin/pkg/plugin/plugin.go index bc256cde..02c6d85b 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,6 +100,16 @@ type IoTDBDataSource struct { // getTablePool on the first table query. tablePoolMu sync.Mutex tablePool *client.TableSessionPool + + // 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 7c115aec..dd7296ba 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. // @@ -310,16 +342,40 @@ func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) b return response } - pool, err := d.getTablePool() + sql, err := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime, qp.IntervalMS) if err != nil { response.Error = err return response } + + dataSet, err := d.executeTableStatement(ctx, qp.Database, sql) + if err != nil { + response.Error = err + return response + } + + response.Frames = append(response.Frames, buildTableResponseFrame(dataSet, qp.Format)) + return response +} + +// 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 + } 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 + return nil, fmt.Errorf("cannot connect to the IoTDB RPC service: %w", err) } defer func() { if closeErr := session.Close(); closeErr != nil { @@ -327,10 +383,9 @@ func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) b } }() - if database := strings.TrimSpace(qp.Database); database != "" { - if err := session.ExecuteNonQueryStatement("USE " + quoteTableIdentifier(database)); err != nil { - response.Error = err - return response + if db := strings.TrimSpace(database); db != "" { + if err := session.ExecuteNonQueryStatement("USE " + quoteTableIdentifier(db)); err != nil { + return nil, err } } @@ -340,15 +395,9 @@ 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 - return response + return nil, err } defer func() { if closeErr := resultSet.Close(); closeErr != nil { @@ -356,14 +405,38 @@ func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) b } }() - 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 { - response.Error = err - return response + return nil, err } + return tableVariableStrings(dataSet) +} - response.Frames = append(response.Frames, buildTableResponseFrame(dataSet, qp.Format)) - return response +// 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 00000000..2d383854 --- /dev/null +++ b/connectors/grafana-plugin/pkg/plugin/table_variable_test.go @@ -0,0 +1,376 @@ +/* + * 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" + "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() + + d := &IoTDBDataSource{} + handler := d.getVariables("Bearer test", http.DefaultClient) + request := httptest.NewRequest(http.MethodGet, "/getVariables?url="+url.QueryEscape(legacy.URL)+"&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) + } +} From 637bb2e079153c3e4c91367be49d2c58b7875d00 Mon Sep 17 00:00:00 2001 From: zxq <3322351820@qq.com> Date: Fri, 21 Aug 2026 17:10:07 +0800 Subject: [PATCH 2/2] Refactor getNodes and request handling to use configured URL, preventing SSRF --- .../pkg/plugin/iotdb_resource_handler.go | 38 ++++++++++++++---- .../pkg/plugin/table_variable_test.go | 39 ++++++++++++++++++- 2 files changed, 67 insertions(+), 10 deletions(-) diff --git a/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go b/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go index dcd4511b..9154ac30 100644 --- a/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go +++ b/connectors/grafana-plugin/pkg/plugin/iotdb_resource_handler.go @@ -43,7 +43,7 @@ func (d *IoTDBDataSource) iotdbResourceHandler(authorization string, httpClient mux := http.NewServeMux() mux.Handle("/getVariables", d.getVariables(authorization, httpClient)) - mux.Handle("/getNodes", getNodes(authorization, httpClient)) + mux.Handle("/getNodes", d.getNodes(authorization, httpClient)) return httpadapter.New(mux) } @@ -53,7 +53,6 @@ type queryReq struct { } type nodeReq struct { Data []string `json:"data"` - Url string `json:"url"` } type queryResp struct { @@ -67,7 +66,6 @@ func (d *IoTDBDataSource) getVariables(authorization string, httpClient *http.Cl http.NotFound(w, r) return } - var url = r.FormValue("url") var sql = r.FormValue("sql") // table:: variables run through the table-model RPC @@ -81,10 +79,22 @@ func (d *IoTDBDataSource) getVariables(authorization string, httpClient *http.Cl 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) @@ -184,7 +194,7 @@ func writeJSONError(w http.ResponseWriter, status int, message string) { _ = json.NewEncoder(w).Encode(queryResp{Code: status, Message: message}) } -func getNodes(authorization string, client *http.Client) http.Handler { +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 { @@ -200,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/table_variable_test.go b/connectors/grafana-plugin/pkg/plugin/table_variable_test.go index 2d383854..b444c32d 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_variable_test.go +++ b/connectors/grafana-plugin/pkg/plugin/table_variable_test.go @@ -22,6 +22,7 @@ import ( "encoding/base64" "encoding/json" "errors" + "io" "net/http" "net/http/httptest" "net/url" @@ -270,9 +271,12 @@ func TestGetVariablesLegacyPath(t *testing.T) { })) defer legacy.Close() - d := &IoTDBDataSource{} + // 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(legacy.URL)+"&sql="+url.QueryEscape("show timeseries"), nil) + 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) @@ -374,3 +378,34 @@ func TestHandleTableVariableQueryExpandsActiveFrom(t *testing.T) { 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) + } +}