diff --git a/connectors/grafana-plugin/README.md b/connectors/grafana-plugin/README.md index 387e5a99..67befbc2 100644 --- a/connectors/grafana-plugin/README.md +++ b/connectors/grafana-plugin/README.md @@ -149,7 +149,7 @@ Unlike the tree-model modes, which go through the REST service, this mode connec DATABASE: the database to run the statement against, picked from the databases the server lists (a name can also be typed, for example a `$database` template variable). 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. -TABLE (optional): picking a table lists its columns grouped by category (TAG, FIELD, ATTRIBUTE). **Use starter query** then replaces the SQL with a query that selects the table's time, TAG and numeric FIELD columns (every FIELD when none is numeric) within the dashboard time range, for example `SELECT "time", "region", "temperature" FROM "weather" WHERE $__timeFilter("time")`. Names are always double-quoted, so reserved words such as `order` and names with spaces stay valid. The table only drives this helper; the SQL is what runs. +TABLE (optional): picking a table lists its columns grouped by category (TAG, FIELD, ATTRIBUTE). **Use starter query** then replaces the SQL with a query that selects the table's time, TAG and numeric FIELD columns (every FIELD when none is numeric) within the dashboard time range, for example `SELECT "time", "region", "temperature" FROM "weather" WHERE $__timeFilter("time")`. Names are always double-quoted, so reserved words such as `order` and names with spaces stay valid. The table only drives this helper; the SQL is what runs. The editor reuses the database, table and column lists it has read for 30 seconds, so a table created in the meantime can take that long to appear (its name can still be typed); opening the DATABASE list always reads the databases again. SQL input box: a single table-model SELECT statement. The following macros are expanded by the plugin before the statement is sent: diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go b/connectors/grafana-plugin/pkg/plugin/plugin.go index 970b7a63..0351c6dd 100644 --- a/connectors/grafana-plugin/pkg/plugin/plugin.go +++ b/connectors/grafana-plugin/pkg/plugin/plugin.go @@ -110,6 +110,11 @@ type IoTDBDataSource struct { // fake to exercise the variable-query path without a live server. tableExecutor func(ctx context.Context, database, sql string) (*tableQueryDataSet, error) + // tableSessionSource hands out a table-model session for the real RPC + // executor. A nil value selects the session pool; tests substitute a fake to + // observe how a statement uses its session without a live server. + tableSessionSource func() (client.ITableSession, 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. diff --git a/connectors/grafana-plugin/pkg/plugin/table_metadata.go b/connectors/grafana-plugin/pkg/plugin/table_metadata.go index 61694567..76dc3330 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_metadata.go +++ b/connectors/grafana-plugin/pkg/plugin/table_metadata.go @@ -20,6 +20,8 @@ package plugin import ( + "context" + "errors" "fmt" "net/http" "strings" @@ -30,6 +32,10 @@ import ( // Resource handlers behind the table-model query editor's database, table and // column pickers. Each one runs a fixed metadata statement on a pooled native // session; request parameters only reach the server as quoted identifiers. +// The statements name their database themselves (SHOW TABLES FROM , +// DESC .), so a lookup issues no USE and changes no session state. +// The request context is honoured as far as the native client allows; see +// executeTableStatement. // tableColumn describes one column of a table-model table as DESC reports it. type tableColumn struct { @@ -60,13 +66,13 @@ func (d *IoTDBDataSource) getTableDatabases() http.Handler { func (d *IoTDBDataSource) getTableTables() http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // The editor trims names when they are entered; here a blank name is only - // rejected, and the name is used as given. + // rejected. The database is trimmed as a panel query's USE trims it. database := r.URL.Query().Get("database") if strings.TrimSpace(database) == "" { writeJSONError(w, http.StatusBadRequest, "the database parameter is required") return } - dataSet, err := d.executeTableStatement(r.Context(), database, "SHOW TABLES") + dataSet, err := d.executeTableStatement(r.Context(), "", showTablesStatement(database)) if err != nil { writeMetadataError(w, "SHOW TABLES", err) return @@ -90,7 +96,7 @@ func (d *IoTDBDataSource) getTableColumns() http.Handler { writeJSONError(w, http.StatusBadRequest, "the database and table parameters are required") return } - dataSet, err := d.executeTableStatement(r.Context(), database, "DESC "+quoteTableIdentifier(table)) + dataSet, err := d.executeTableStatement(r.Context(), "", describeTableStatement(database, table)) if err != nil { writeMetadataError(w, "DESC", err) return @@ -104,8 +110,25 @@ func (d *IoTDBDataSource) getTableColumns() http.Handler { }) } +// showTablesStatement lists the tables of a database without a USE. +func showTablesStatement(database string) string { + return "SHOW TABLES FROM " + quoteTableIdentifier(strings.TrimSpace(database)) +} + +// describeTableStatement describes a table by its qualified name, without a +// USE. The table name is used as given: the editor trims it when it is +// entered, and the starter query names the same table. +func describeTableStatement(database, table string) string { + return "DESC " + quoteTableIdentifier(strings.TrimSpace(database)) + "." + quoteTableIdentifier(table) +} + func writeMetadataError(w http.ResponseWriter, statement string, err error) { - log.DefaultLogger.Error("table-model metadata query failed", "statement", statement, "err", err) + if errors.Is(err, context.Canceled) { + // The editor cancels a lookup it no longer needs; that is not a failure. + log.DefaultLogger.Debug("table-model metadata query cancelled", "statement", statement) + } else { + log.DefaultLogger.Error("table-model metadata query failed", "statement", statement, "err", err) + } writeJSONError(w, http.StatusInternalServerError, err.Error()) } diff --git a/connectors/grafana-plugin/pkg/plugin/table_metadata_test.go b/connectors/grafana-plugin/pkg/plugin/table_metadata_test.go index b3f869ce..b310c47d 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_metadata_test.go +++ b/connectors/grafana-plugin/pkg/plugin/table_metadata_test.go @@ -28,6 +28,8 @@ import ( "reflect" "strings" "testing" + + "github.com/apache/iotdb-client-go/v2/client" ) type recordedStatement struct { @@ -80,7 +82,7 @@ func TestTableDatabasesReturnsTheDatabaseColumn(t *testing.T) { } } -func TestTableTablesRunsShowTablesInTheRequestedDatabase(t *testing.T) { +func TestTableTablesNamesTheRequestedDatabaseInTheStatement(t *testing.T) { var calls []recordedStatement d := metadataDataSource(&tableQueryDataSet{ ColumnNames: []string{"TableName", "TTL(ms)"}, @@ -99,7 +101,8 @@ func TestTableTablesRunsShowTablesInTheRequestedDatabase(t *testing.T) { if want := []string{"weather station", "cpu"}; !reflect.DeepEqual(names, want) { t.Fatalf("names = %v, want %v", names, want) } - if want := []recordedStatement{{database: "grafana_demo", sql: "SHOW TABLES"}}; !reflect.DeepEqual(calls, want) { + // SHOW TABLES FROM names the database, so no USE is needed on the session. + if want := []recordedStatement{{database: "", sql: `SHOW TABLES FROM "grafana_demo"`}}; !reflect.DeepEqual(calls, want) { t.Fatalf("statements = %v, want %v", calls, want) } } @@ -134,12 +137,61 @@ func TestTableColumnsQuotesTheTableName(t *testing.T) { if !reflect.DeepEqual(columns, want) { t.Fatalf("columns = %v, want %v", columns, want) } - // An embedded double quote is doubled, so the name stays one identifier. - if want := []recordedStatement{{database: "grafana_demo", sql: `DESC "weather""station"`}}; !reflect.DeepEqual(calls, want) { + // An embedded double quote is doubled, so the name stays one identifier, and + // the table is qualified with its database instead of a USE. + if want := []recordedStatement{{database: "", sql: `DESC "grafana_demo"."weather""station"`}}; !reflect.DeepEqual(calls, want) { t.Fatalf("statements = %v, want %v", calls, want) } } +func TestTableMetadataQuotesEachPartOfTheName(t *testing.T) { + for _, tc := range []struct { + name string + handler func(d *IoTDBDataSource) http.Handler + target string + want string + }{ + { + name: "hyphen and upper case in the database", + handler: (*IoTDBDataSource).getTableTables, + target: "/tableTables?database=grafana-Demo", + want: `SHOW TABLES FROM "grafana-Demo"`, + }, + { + name: "double quote in the database", + handler: (*IoTDBDataSource).getTableTables, + target: `/tableTables?database=we%22ird`, + want: `SHOW TABLES FROM "we""ird"`, + }, + { + name: "hyphen and upper case in both parts", + handler: (*IoTDBDataSource).getTableColumns, + target: "/tableColumns?database=grafana-Demo&table=Weather-Station", + want: `DESC "grafana-Demo"."Weather-Station"`, + }, + { + // A dot inside a name stays inside its quoted part, so it cannot + // redirect the lookup to another database. + name: "dot and double quote inside the names", + handler: (*IoTDBDataSource).getTableColumns, + target: `/tableColumns?database=a.b&table=c%22.%22d`, + want: `DESC "a.b"."c"".""d"`, + }, + } { + t.Run(tc.name, func(t *testing.T) { + var calls []recordedStatement + d := metadataDataSource(&tableQueryDataSet{ColumnNames: []string{"TableName", "ColumnName", "DataType", "Category"}}, nil, &calls) + + if recorder := serveMetadata(t, tc.handler(d), tc.target); recorder.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) + } + if want := []recordedStatement{{database: "", sql: tc.want}}; !reflect.DeepEqual(calls, want) { + t.Fatalf("statements = %v, want %v", calls, want) + } + }) + } +} + func TestTableColumnsUsesTheNamesAsGiven(t *testing.T) { var calls []recordedStatement d := metadataDataSource(&tableQueryDataSet{ @@ -147,14 +199,15 @@ func TestTableColumnsUsesTheNamesAsGiven(t *testing.T) { Values: [][]interface{}{{"time", "TIMESTAMP", "TIME"}}, }, nil, &calls) - recorder := serveMetadata(t, d.getTableColumns(), `/tableColumns?database=grafana_demo&table=weather%20`) + recorder := serveMetadata(t, d.getTableColumns(), `/tableColumns?database=%20grafana_demo%20&table=weather%20`) if recorder.Code != http.StatusOK { t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) } // The editor trims names when they are entered, so the handler does not trim - // again: the columns it lists are those of the table the starter query names. - if want := []recordedStatement{{database: "grafana_demo", sql: `DESC "weather "`}}; !reflect.DeepEqual(calls, want) { + // the table again: the columns it lists are those of the table the starter + // query names. The database is trimmed, as a panel query's USE trims it. + if want := []recordedStatement{{database: "", sql: `DESC "grafana_demo"."weather "`}}; !reflect.DeepEqual(calls, want) { t.Fatalf("statements = %v, want %v", calls, want) } } @@ -181,6 +234,33 @@ func TestTableMetadataRequiresItsParameters(t *testing.T) { } } +func TestTableMetadataStopsWhenTheEditorCancelsTheRequest(t *testing.T) { + acquired := 0 + d := sessionDataSource(&acquired, func() client.ITableSession { return &fakeTableSession{} }) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + for _, tc := range []struct { + handler http.Handler + target string + }{ + {d.getTableDatabases(), "/tableDatabases"}, + {d.getTableTables(), "/tableTables?database=grafana_demo"}, + {d.getTableColumns(), "/tableColumns?database=grafana_demo&table=cpu"}, + } { + recorder := httptest.NewRecorder() + tc.handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, tc.target, nil).WithContext(ctx)) + if recorder.Code != http.StatusInternalServerError { + t.Errorf("%s: status = %d, want %d", tc.target, recorder.Code, http.StatusInternalServerError) + } + } + // The handlers pass the request's context on, so an aborted lookup never + // takes one of the pool's sessions from the panel queries. + if acquired != 0 { + t.Fatalf("cancelled lookups took %d session(s)", acquired) + } +} + func TestTableMetadataReportsServerErrors(t *testing.T) { var calls []recordedStatement d := metadataDataSource(nil, errors.New("550: Table 'grafana_demo.nope' does not exist"), &calls) diff --git a/connectors/grafana-plugin/pkg/plugin/table_query.go b/connectors/grafana-plugin/pkg/plugin/table_query.go index b42ede13..a937a515 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_query.go +++ b/connectors/grafana-plugin/pkg/plugin/table_query.go @@ -245,7 +245,8 @@ func formatIoTDBDuration(intervalMS int64) (string, error) { // 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. +// form, so a database or table name reaches the server verbatim. Each part of +// a qualified name is quoted on its own ("db"."table"). func quoteTableIdentifier(name string) string { return `"` + strings.ReplaceAll(name, `"`, `""`) + `"` } @@ -521,19 +522,34 @@ func instantSeriesKey(dataSet *tableQueryDataSet, row []interface{}, timeCol int // 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. +// A non-empty database is USEd on the session so a statement's table +// references resolve against it regardless of any session state left by +// earlier queries; a statement that names its database itself passes an empty +// one and issues no USE. When tableExecutor is non-nil it is used instead of +// the real RPC path. +// +// The native client's pool takes no context: GetSession can wait up to +// tableWaitSessionTimeoutMs for a free session and cannot be interrupted. The +// context is therefore checked before that wait and again after it, so a +// request that has been cancelled, or whose deadline has passed, never runs a +// statement whose result nobody will read. 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 { + if err := ctx.Err(); err != nil { return nil, err } - session, err := pool.GetSession() + getSession := d.tableSessionSource + if getSession == nil { + pool, err := d.getTablePool() + if err != nil { + return nil, err + } + getSession = pool.GetSession + } + session, err := getSession() if err != nil { log.DefaultLogger.Error("Cannot connect to the IoTDB RPC service", "err", err) return nil, fmt.Errorf("cannot connect to the IoTDB RPC service: %w", err) @@ -543,6 +559,11 @@ func (d *IoTDBDataSource) executeTableStatement(ctx context.Context, database, s log.DefaultLogger.Error("Failed to return session to the pool", "err", closeErr) } }() + if err := ctx.Err(); err != nil { + // The caller gave up while this request waited for a session; hand the + // session straight back to the pool for the requests still waiting. + return nil, err + } if db := strings.TrimSpace(database); db != "" { if err := session.ExecuteNonQueryStatement("USE " + quoteTableIdentifier(db)); err != nil { diff --git a/connectors/grafana-plugin/pkg/plugin/table_query_test.go b/connectors/grafana-plugin/pkg/plugin/table_query_test.go index 44d97eee..97c167c5 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_query_test.go +++ b/connectors/grafana-plugin/pkg/plugin/table_query_test.go @@ -21,10 +21,12 @@ import ( "context" "encoding/json" "errors" + "reflect" "strings" "testing" "time" + "github.com/apache/iotdb-client-go/v2/client" "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/data" ) @@ -320,6 +322,126 @@ func TestFetchTableDataSetPropagatesError(t *testing.T) { } } +// errQueryNotServed ends a statement on fakeTableSession: a *client.SessionDataSet +// cannot be built without a server, so the tests only look at which +// statements reached the session. +var errQueryNotServed = errors.New("fake session serves no results") + +// fakeTableSession stands in for a pooled native-client table session and +// records the statements run on it. +type fakeTableSession struct { + statements []string + closed bool +} + +func (s *fakeTableSession) Insert(*client.Tablet) error { return errors.New("not supported") } + +func (s *fakeTableSession) ExecuteNonQueryStatement(sql string) error { + s.statements = append(s.statements, sql) + return nil +} + +func (s *fakeTableSession) ExecuteQueryStatement(sql string, _ *int64) (*client.SessionDataSet, error) { + s.statements = append(s.statements, sql) + return nil, errQueryNotServed +} + +func (s *fakeTableSession) Close() error { + s.closed = true + return nil +} + +// sessionDataSource returns a datasource whose real RPC executor takes its +// sessions from source, counting each acquisition. +func sessionDataSource(acquired *int, source func() client.ITableSession) *IoTDBDataSource { + return &IoTDBDataSource{ + tableSessionSource: func() (client.ITableSession, error) { + *acquired++ + return source(), nil + }, + } +} + +func TestTableStatementUsesTheDatabaseOnlyWhenOneIsGiven(t *testing.T) { + for _, tc := range []struct { + name string + database string + sql string + want []string + }{ + // A panel or variable query keeps its USE, as before. + {name: "panel query", database: " grafana_demo ", sql: "SELECT 1", want: []string{`USE "grafana_demo"`, "SELECT 1"}}, + // A metadata statement names its database itself and needs no USE. + {name: "metadata lookup", database: "", sql: `SHOW TABLES FROM "grafana_demo"`, want: []string{`SHOW TABLES FROM "grafana_demo"`}}, + } { + t.Run(tc.name, func(t *testing.T) { + acquired := 0 + session := &fakeTableSession{} + d := sessionDataSource(&acquired, func() client.ITableSession { return session }) + + if _, err := d.executeTableStatement(context.Background(), tc.database, tc.sql); !errors.Is(err, errQueryNotServed) { + t.Fatalf("error = %v, want %v", err, errQueryNotServed) + } + if !reflect.DeepEqual(session.statements, tc.want) { + t.Fatalf("statements = %q, want %q", session.statements, tc.want) + } + if !session.closed { + t.Fatalf("the session was not handed back to the pool") + } + }) + } +} + +func TestTableStatementTakesNoSessionOnceTheRequestIsOver(t *testing.T) { + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + expired, cancelExpired := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + defer cancelExpired() + + for _, tc := range []struct { + name string + ctx context.Context + want error + }{ + {name: "cancelled", ctx: cancelled, want: context.Canceled}, + {name: "deadline passed", ctx: expired, want: context.DeadlineExceeded}, + } { + t.Run(tc.name, func(t *testing.T) { + acquired := 0 + d := sessionDataSource(&acquired, func() client.ITableSession { return &fakeTableSession{} }) + + if _, err := d.executeTableStatement(tc.ctx, "", `SHOW TABLES FROM "grafana_demo"`); !errors.Is(err, tc.want) { + t.Fatalf("error = %v, want %v", err, tc.want) + } + if acquired != 0 { + t.Fatalf("a request that was already over waited for %d session(s)", acquired) + } + }) + } +} + +func TestTableStatementHandsBackASessionThatArrivesAfterTheRequestEnded(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + acquired := 0 + session := &fakeTableSession{} + d := sessionDataSource(&acquired, func() client.ITableSession { + // GetSession cannot be interrupted, so the caller gives up while it waits. + cancel() + return session + }) + + if _, err := d.executeTableStatement(ctx, "grafana_demo", "SELECT 1"); !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want %v", err, context.Canceled) + } + if len(session.statements) != 0 { + t.Fatalf("statements = %q, want none once the request has ended", session.statements) + } + if !session.closed { + t.Fatalf("the session was not handed back to the pool") + } +} + // 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. diff --git a/connectors/grafana-plugin/src/componments/TablePicker.test.tsx b/connectors/grafana-plugin/src/componments/TablePicker.test.tsx index 9e6934f9..77084d6e 100644 --- a/connectors/grafana-plugin/src/componments/TablePicker.test.tsx +++ b/connectors/grafana-plugin/src/componments/TablePicker.test.tsx @@ -58,12 +58,64 @@ describe('TablePicker', () => { const variables: Record = { $db: 'db1' }; const ds = fakeDatasource(variables); const { rerender, props } = renderPicker(ds, { database: '$db' }); - await waitFor(() => expect(ds.getTableTables).toHaveBeenCalledWith('db1')); + await waitFor(() => expect(ds.getTableTables).toHaveBeenCalledWith('db1', expect.anything())); variables.$db = 'db2'; rerender(); - await waitFor(() => expect(ds.getTableTables).toHaveBeenCalledWith('db2')); + await waitFor(() => expect(ds.getTableTables).toHaveBeenCalledWith('db2', expect.anything())); + }); + + it('cancels a lookup that is still running when the picker unmounts', async () => { + const ds = fakeDatasource({}); + ds.getTableTables.mockReturnValue(new Promise(() => {})); + const { unmount } = renderPicker(ds, { database: 'db1' }); + await waitFor(() => expect(ds.getTableTables).toHaveBeenCalledTimes(1)); + const { signal } = ds.getTableTables.mock.calls[0][1]; + expect(signal.aborted).toBe(false); + + unmount(); + + expect(signal.aborted).toBe(true); + }); + + it('cancels the lookup for a database that is no longer selected, without reporting it', async () => { + const variables: Record = { $db: 'db1' }; + const ds = fakeDatasource(variables); + // Like a real request, the db1 lookup fails once it is aborted. + ds.getTableTables.mockImplementation((database: string, { signal }: { signal: AbortSignal }) => + database === 'db1' + ? new Promise((_, reject) => signal.addEventListener('abort', () => reject(new Error('request aborted')))) + : Promise.resolve(['cpu']) + ); + const { rerender, props } = renderPicker(ds, { database: '$db' }); + await waitFor(() => expect(ds.getTableTables).toHaveBeenCalledWith('db1', expect.anything())); + + variables.$db = 'db2'; + rerender(); + await waitFor(() => expect(ds.getTableTables).toHaveBeenCalledWith('db2', expect.anything())); + await act(async () => {}); + + const [first, second] = ds.getTableTables.mock.calls.map(([, options]) => options.signal); + expect(first.aborted).toBe(true); + expect(second.aborted).toBe(false); + expect(screen.queryByText(/request aborted/)).toBeNull(); + }); + + it('asks past the datasource cache only when the database list is opened', async () => { + const ds = fakeDatasource({}); + renderPicker(ds, {}); + await waitFor(() => expect(ds.getTableDatabases).toHaveBeenCalledTimes(1)); + expect(ds.getTableDatabases.mock.calls[0][0]).toMatchObject({ refresh: false }); + const databaseInput = document.getElementById('iotdb-table-database-A') as HTMLInputElement; + + await act(async () => { + fireEvent.focus(databaseInput); + fireEvent.keyDown(databaseInput, { key: 'ArrowDown' }); + }); + + await waitFor(() => expect(ds.getTableDatabases).toHaveBeenCalledTimes(2)); + expect(ds.getTableDatabases.mock.calls[1][0]).toMatchObject({ refresh: true }); }); it('keeps a typed database name when the picker loses focus', async () => { diff --git a/connectors/grafana-plugin/src/componments/TablePicker.tsx b/connectors/grafana-plugin/src/componments/TablePicker.tsx index 6ad6333c..0f00471b 100644 --- a/connectors/grafana-plugin/src/componments/TablePicker.tsx +++ b/connectors/grafana-plugin/src/componments/TablePicker.tsx @@ -79,31 +79,35 @@ function errorMessage(err: unknown): string { // useLoaded runs one metadata lookup whenever its key changes. State is only // written when a lookup settles, and a result is only used while its key is -// current. A failed lookup keeps its message; the pickers still accept a typed -// name, so the editor stays usable when metadata cannot be read. With -// keepPrevious, the last value stays available while a new lookup is running. -function useLoaded(key: string, load: (() => Promise) | undefined, keepPrevious = false): Loaded { +// current: when the key changes or the picker unmounts, the lookup's signal is +// aborted, which cancels its request. A failed lookup keeps its message; the +// pickers still accept a typed name, so the editor stays usable when metadata +// cannot be read. With keepPrevious, the last value stays available while a +// new lookup is running. +function useLoaded( + key: string, + load: ((signal: AbortSignal) => Promise) | undefined, + keepPrevious = false +): Loaded { const [result, setResult] = useState<{ key: string; value?: T; error?: string }>(); useEffect(() => { if (!load) { return; } - let cancelled = false; - load().then( + const controller = new AbortController(); + load(controller.signal).then( (value) => { - if (!cancelled) { + if (!controller.signal.aborted) { setResult({ key, value }); } }, (err) => { - if (!cancelled) { + if (!controller.signal.aborted) { setResult({ key, error: errorMessage(err) }); } } ); - return () => { - cancelled = true; - }; + return () => controller.abort(); // The key identifies the lookup; `load` is a fresh closure on every render. // eslint-disable-next-line react-hooks/exhaustive-deps }, [key]); @@ -157,17 +161,24 @@ export function TablePicker({ // changes value fetches the new database's tables and columns. const resolvedDatabase = database ? datasource.resolveTemplate(database, scopedVars) : ''; const resolvedTable = table ? datasource.resolveTemplate(table, scopedVars) : ''; - // Opening the database list asks the server again, so a database created - // later, or a lookup that failed, is picked up without reopening the editor. + // Opening the database list asks the server again, past the datasource's + // cache, so a database created later, or a lookup that failed, is picked up + // without reopening the editor. const [databaseRefresh, setDatabaseRefresh] = useState(0); - const databases = useLoaded(`databases|${databaseRefresh}`, () => datasource.getTableDatabases(), true); + const databases = useLoaded( + `databases|${databaseRefresh}`, + (signal) => datasource.getTableDatabases({ signal, refresh: databaseRefresh > 0 }), + true + ); const tables = useLoaded( `tables|${resolvedDatabase}`, - resolvedDatabase ? () => datasource.getTableTables(resolvedDatabase) : undefined + resolvedDatabase ? (signal) => datasource.getTableTables(resolvedDatabase, { signal }) : undefined ); const columns = useLoaded( `columns|${resolvedDatabase}|${resolvedTable}`, - resolvedDatabase && resolvedTable ? () => datasource.getTableColumns(resolvedDatabase, resolvedTable) : undefined + resolvedDatabase && resolvedTable + ? (signal) => datasource.getTableColumns(resolvedDatabase, resolvedTable, { signal }) + : undefined ); const errors = Array.from(new Set([databases.error, tables.error, columns.error].filter(Boolean))); diff --git a/connectors/grafana-plugin/src/datasource.test.ts b/connectors/grafana-plugin/src/datasource.test.ts index 0faa85c1..b892e6ec 100644 --- a/connectors/grafana-plugin/src/datasource.test.ts +++ b/connectors/grafana-plugin/src/datasource.test.ts @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { DataSource } from './datasource'; +import { DataSource, tableMetadataTtlMs } from './datasource'; import { applyQueryType, getQueryType, IoTDBQuery } from './types'; import { ScopedVars } from '@grafana/data'; @@ -303,10 +303,27 @@ describe('query type model', () => { }); describe('table-model metadata lookups', () => { - it('do not raise a global error alert, since the editor shows failures inline', async () => { + let now: number; + let clock: jest.SpyInstance; + + beforeEach(() => { + now = 1_000_000; + clock = jest.spyOn(Date, 'now').mockImplementation(() => now); + }); + + afterEach(() => { + clock.mockRestore(); + }); + + function metadataDataSource() { const ds = new DataSource({ jsonData: { url: 'http://localhost:6667', username: 'root' } } as any); const getResource = jest.fn().mockResolvedValue([]); (ds as unknown as { getResource: jest.Mock }).getResource = getResource; + return { ds, getResource }; + } + + it('do not raise a global error alert, since the editor shows failures inline', async () => { + const { ds, getResource } = metadataDataSource(); await ds.getTableDatabases(); await ds.getTableTables('db'); await ds.getTableColumns('db', 't'); @@ -316,4 +333,78 @@ describe('table-model metadata lookups', () => { ['tableColumns', { database: 'db', table: 't' }, { showErrorAlert: false }], ]); }); + + it('reuse a recent result for the same database and table', async () => { + const { ds, getResource } = metadataDataSource(); + getResource.mockImplementation((path: string, params?: Record) => + Promise.resolve([`${path} ${JSON.stringify(params ?? {})}`]) + ); + + expect(await ds.getTableTables('db1')).toEqual(['tableTables {"database":"db1"}']); + expect(await ds.getTableTables('db1')).toEqual(['tableTables {"database":"db1"}']); + await ds.getTableTables('db2'); + await ds.getTableColumns('db1', 't'); + await ds.getTableColumns('db1', 't'); + await ds.getTableColumns('db1', 'u'); + await ds.getTableColumns('db2', 't'); + await ds.getTableDatabases(); + await ds.getTableDatabases(); + + expect(getResource.mock.calls.map(([path, params]) => [path, params])).toEqual([ + ['tableTables', { database: 'db1' }], + ['tableTables', { database: 'db2' }], + ['tableColumns', { database: 'db1', table: 't' }], + ['tableColumns', { database: 'db1', table: 'u' }], + ['tableColumns', { database: 'db2', table: 't' }], + ['tableDatabases', undefined], + ]); + }); + + it('ask the server again once the cached result has expired', async () => { + const { ds, getResource } = metadataDataSource(); + getResource.mockResolvedValueOnce(['cpu']).mockResolvedValueOnce(['cpu', 'mem']); + + await ds.getTableTables('db1'); + now += tableMetadataTtlMs - 1; + expect(await ds.getTableTables('db1')).toEqual(['cpu']); + expect(getResource).toHaveBeenCalledTimes(1); + + now += 1; + expect(await ds.getTableTables('db1')).toEqual(['cpu', 'mem']); + expect(getResource).toHaveBeenCalledTimes(2); + }); + + it('do not keep a failed lookup', async () => { + const { ds, getResource } = metadataDataSource(); + getResource.mockRejectedValueOnce(new Error('Unknown database db1')).mockResolvedValueOnce(['cpu']); + + await expect(ds.getTableTables('db1')).rejects.toThrow('Unknown database db1'); + expect(await ds.getTableTables('db1')).toEqual(['cpu']); + expect(getResource).toHaveBeenCalledTimes(2); + }); + + it('ask the server past the cache on refresh, and keep the fresh answer', async () => { + const { ds, getResource } = metadataDataSource(); + getResource.mockResolvedValueOnce(['db1']).mockResolvedValueOnce(['db1', 'db2']); + + await ds.getTableDatabases(); + expect(await ds.getTableDatabases({ refresh: true })).toEqual(['db1', 'db2']); + expect(await ds.getTableDatabases()).toEqual(['db1', 'db2']); + expect(getResource).toHaveBeenCalledTimes(2); + }); + + it("pass the caller's abort signal to the request", async () => { + const { ds, getResource } = metadataDataSource(); + const controller = new AbortController(); + + await ds.getTableColumns('db1', 't', { signal: controller.signal }); + + expect(getResource).toHaveBeenCalledWith( + 'tableColumns', + { database: 'db1', table: 't' }, + { showErrorAlert: false, abortSignal: controller.signal } + ); + // Equality alone would accept any AbortSignal; it must be the caller's. + expect(getResource.mock.calls[0][2].abortSignal).toBe(controller.signal); + }); }); diff --git a/connectors/grafana-plugin/src/datasource.ts b/connectors/grafana-plugin/src/datasource.ts index aff96a40..37c67d7e 100644 --- a/connectors/grafana-plugin/src/datasource.ts +++ b/connectors/grafana-plugin/src/datasource.ts @@ -20,6 +20,18 @@ import { IoTDBOptions, IoTDBQuery, TableColumn } from './types'; import { toMetricFindValue } from './functions'; import { DataSourceWithBackend, getTemplateSrv } from '@grafana/runtime'; +// How long a table-model metadata lookup is reused. Short, so that a table +// created meanwhile shows up soon; opening the database list always asks the +// server again (see TableMetadataOptions.refresh). +export const tableMetadataTtlMs = 30000; + +export interface TableMetadataOptions { + // Cancels the request when the caller no longer needs its result. + signal?: AbortSignal; + // Asks the server even when a recent result is cached, and caches the answer. + refresh?: boolean; +} + function variableQueryText(query: unknown): string | undefined { if (typeof query === 'string') { return query; @@ -190,16 +202,44 @@ export class DataSource extends DataSourceWithBackend // Metadata behind the table-model editor's pickers. Callers pass names that // resolveTemplate has already expanded. The editor shows a failed lookup // inline, so it is not also raised as a global error alert. - getTableDatabases(): Promise { - return this.getResource('tableDatabases', undefined, { showErrorAlert: false }); + getTableDatabases(options?: TableMetadataOptions): Promise { + return this.tableMetadata('tableDatabases', {}, options); + } + + getTableTables(database: string, options?: TableMetadataOptions): Promise { + return this.tableMetadata('tableTables', { database }, options); } - getTableTables(database: string): Promise { - return this.getResource('tableTables', { database }, { showErrorAlert: false }); + getTableColumns(database: string, table: string, options?: TableMetadataOptions): Promise { + return this.tableMetadata('tableColumns', { database, table }, options); } - getTableColumns(database: string, table: string): Promise { - return this.getResource('tableColumns', { database, table }, { showErrorAlert: false }); + // A successful lookup is kept for tableMetadataTtlMs, keyed on the endpoint + // and the names it was asked about, so that opening the editor again or + // switching between queries does not ask the server for names it has just + // listed. A failed lookup is not kept, so the next one asks again. + private tableMetadataCache = new Map(); + + private async tableMetadata( + path: string, + names: Record, + { signal, refresh = false }: TableMetadataOptions = {} + ): Promise { + const key = JSON.stringify([path, names]); + const cached = this.tableMetadataCache.get(key); + if (cached && !refresh && cached.expires > Date.now()) { + return cached.value as T; + } + const params = Object.keys(names).length > 0 ? names : undefined; + const value = await this.getResource(path, params, { showErrorAlert: false, abortSignal: signal }); + const now = Date.now(); + for (const [staleKey, entry] of this.tableMetadataCache) { + if (entry.expires <= now) { + this.tableMetadataCache.delete(staleKey); + } + } + this.tableMetadataCache.set(key, { expires: now + tableMetadataTtlMs, value }); + return value; } async getVariablesResult(sql: string) {