support dynamic IoTDB table-model variables

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,30 +53,48 @@ type queryReq struct {
 }
 type nodeReq struct {
 	Data []string `json:"data"`
-	Url string `json:"url"`
 }
 
 type queryResp struct {
 	Code int `json:"code"`
 	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)

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

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,