From f4fc56a3a2756f2ea95e090c9223fabefc45ace6 Mon Sep 17 00:00:00 2001 From: Zihan Dai <1436286758@qq.com> Date: Tue, 20 Oct 2026 22:04:34 +1100 Subject: [PATCH] Cache, cancel and simplify the Grafana table-model metadata lookups Follow-up to the review of #134. The table-model editor's database, table and column pickers share the plugin's eight-session pool with panel queries. Each lookup ran again on every editor mount, could not be cancelled, and issued an extra USE. - The lookups now run SHOW TABLES FROM and DESC ., with each part of the name quoted on its own, so they issue no USE and leave no session state behind. Panel and variable queries still USE their database. - The table-model executor checks the request context before it waits for a pooled session and again once it has one. The native client's GetSession takes no context and cannot be interrupted, so a request cancelled during the wait still waits, but it then hands the session straight back instead of running its statement. - The datasource keeps successful lookups for 30 seconds, keyed on the database and table. Opening the database list still asks the server again, and failed lookups are not kept. - The pickers pass an AbortSignal with each lookup and abort it when the picker unmounts or its database or table changes, which cancels the request. Signed-off-by: Zihan Dai <1436286758@qq.com> --- connectors/grafana-plugin/README.md | 2 +- .../grafana-plugin/pkg/plugin/plugin.go | 5 + .../pkg/plugin/table_metadata.go | 31 ++++- .../pkg/plugin/table_metadata_test.go | 94 +++++++++++++- .../grafana-plugin/pkg/plugin/table_query.go | 35 ++++- .../pkg/plugin/table_query_test.go | 122 ++++++++++++++++++ .../src/componments/TablePicker.test.tsx | 56 +++++++- .../src/componments/TablePicker.tsx | 43 +++--- .../grafana-plugin/src/datasource.test.ts | 95 +++++++++++++- connectors/grafana-plugin/src/datasource.ts | 52 +++++++- 10 files changed, 490 insertions(+), 45 deletions(-) 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) {