From 4b95d47923a566a3b545031a6d611818f6a9fc9f Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Thu, 17 Jun 2021 08:53:52 +0530 Subject: [PATCH] DB driver implementation via RPC (#17779) This PR builds up on the pass-through DB driver to a fully functioning DB driver implementation via our RPC layer. To keep things separate from the plugin RPC API, and have the ability to move fast with changes, a separate field Driver is added to MattermostPlugin. Typically the field which is required to be compatible are the API and Helpers. It would be well-documented that Driver is purely for internal use by Mattermost plugins. A new Driver interface was created which would have a client and server implementation. Every object (connection, statement, etc.) is created and added to a map on the server side. On the client side, the wrapper structs hold the object id, and communicate via the RPC API using this id. When the server gets the object id, it picks up the appropriate object from its map and performs the operation, and sends back the data. Some things that need to be handled are errors. Typical error types like pq.Error and mysql.MySQLError are registered with encoding/gob. But for error variables like sql.ErrNoRows, a special integer is encoded with the ErrorString struct. And on the cilent side, the integer is checked, and the appropriate error variable is returned. Some pending things: - Context support. This is tricky. Since context.Context is an interface, it's not possible to marshal it. We have to find a way to get the timeout value from the context and pass it. - RowsColumnScanType(rowsID string, index int) reflect.Type API. Again, reflect.Type is an interface. - Master/Replica API support. --- app/plugin.go | 2 +- app/plugin_api_test.go | 24 +- app/plugin_api_tests/test_db_driver/main.go | 57 +++ app/plugin_db_driver.go | 303 +++++++++++++ app/plugin_hooks_test.go | 4 +- plugin/client.go | 7 + plugin/client_rpc.go | 97 +++- plugin/db_rpc.go | 461 ++++++++++++++++++++ plugin/environment.go | 10 +- plugin/health_check_test.go | 4 +- plugin/helpers.go | 54 +++ plugin/supervisor.go | 7 +- plugin/supervisor_test.go | 6 +- services/telemetry/telemetry_test.go | 7 +- shared/driver/conn.go | 121 ++--- shared/driver/driver.go | 32 +- shared/driver/objects.go | 80 ++-- store/sqlstore/store_test.go | 43 -- utils/test_files_compiler.go | 29 ++ web/web_test.go | 2 +- 20 files changed, 1176 insertions(+), 174 deletions(-) create mode 100644 app/plugin_api_tests/test_db_driver/main.go create mode 100644 app/plugin_db_driver.go create mode 100644 plugin/db_rpc.go diff --git a/app/plugin.go b/app/plugin.go index bba8b1eaa0..0937c707c3 100644 --- a/app/plugin.go +++ b/app/plugin.go @@ -199,7 +199,7 @@ func (s *Server) initPlugins(c *request.Context, pluginDir, webappPluginDir stri return New(ServerConnector(s)).NewPluginAPI(c, manifest) } - env, err := plugin.NewEnvironment(newApiFunc, pluginDir, webappPluginDir, s.Log, s.Metrics) + env, err := plugin.NewEnvironment(newApiFunc, NewDriverImpl(s), pluginDir, webappPluginDir, s.Log, s.Metrics) if err != nil { mlog.Error("Failed to start up plugins", mlog.Err(err)) return diff --git a/app/plugin_api_test.go b/app/plugin_api_test.go index 3a77edd8e2..4a81a6c83a 100644 --- a/app/plugin_api_test.go +++ b/app/plugin_api_test.go @@ -69,7 +69,7 @@ func setDefaultPluginConfig(th *TestHelper, pluginID string) { }) } -func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests []string, pluginIDs []string, app *App, c *request.Context) string { +func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests []string, pluginIDs []string, asMain bool, app *App, c *request.Context) string { pluginDir, err := ioutil.TempDir("", "") require.NoError(t, err) t.Cleanup(func() { @@ -92,7 +92,7 @@ func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests return app.NewPluginAPI(c, manifest) } - env, err := plugin.NewEnvironment(newPluginAPI, pluginDir, webappPluginDir, app.Log(), nil) + env, err := plugin.NewEnvironment(newPluginAPI, NewDriverImpl(app.srv), pluginDir, webappPluginDir, app.Log(), nil) require.NoError(t, err) require.Equal(t, len(pluginCodes), len(pluginIDs)) @@ -100,7 +100,11 @@ func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests for i, pluginID := range pluginIDs { backend := filepath.Join(pluginDir, pluginID, "backend.exe") - utils.CompileGo(t, pluginCodes[i], backend) + if asMain { + utils.CompileGo(t, pluginCodes[i], backend) + } else { + utils.CompileGoTest(t, pluginCodes[i], backend) + } ioutil.WriteFile(filepath.Join(pluginDir, pluginID, "plugin.json"), []byte(pluginManifests[i]), 0600) manifest, activated, reterr := env.Activate(pluginID) @@ -121,7 +125,11 @@ func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests } func setupPluginApiTest(t *testing.T, pluginCode string, pluginManifest string, pluginID string, app *App, c *request.Context) string { - return setupMultiPluginApiTest(t, []string{pluginCode}, []string{pluginManifest}, []string{pluginID}, app, c) + + asMain := pluginID != "test_db_driver" + return setupMultiPluginApiTest(t, + []string{pluginCode}, []string{pluginManifest}, []string{pluginID}, + asMain, app, c) } func TestPublicFilesPathConfiguration(t *testing.T) { @@ -766,7 +774,7 @@ func TestPluginAPIGetPlugins(t *testing.T) { defer os.RemoveAll(pluginDir) defer os.RemoveAll(webappPluginDir) - env, err := plugin.NewEnvironment(th.NewPluginAPI, pluginDir, webappPluginDir, th.App.Log(), nil) + env, err := plugin.NewEnvironment(th.NewPluginAPI, NewDriverImpl(th.Server), pluginDir, webappPluginDir, th.App.Log(), nil) require.NoError(t, err) pluginIDs := []string{"pluginid1", "pluginid2", "pluginid3"} @@ -854,7 +862,7 @@ func TestInstallPlugin(t *testing.T) { return app.NewPluginAPI(c, manifest) } - env, err := plugin.NewEnvironment(newPluginAPI, pluginDir, webappPluginDir, app.Log(), nil) + env, err := plugin.NewEnvironment(newPluginAPI, NewDriverImpl(app.srv), pluginDir, webappPluginDir, app.Log(), nil) require.NoError(t, err) app.SetPluginsEnvironment(env) @@ -1053,6 +1061,7 @@ func pluginAPIHookTest(t *testing.T, th *TestHelper, fileName string, id string, if settingsSchema != "" { schema = settingsSchema } + th.App.srv.sqlStore = th.GetSqlStore() setupPluginApiTest(t, code, fmt.Sprintf(`{"id": "%v", "backend": {"executable": "backend.exe"}, "settings_schema": %v}`, id, schema), id, th.App, th.Context) @@ -1481,6 +1490,7 @@ func TestInterpluginPluginHTTP(t *testing.T) { "testplugininterserver", "testplugininterclient", }, + true, th.App, th.Context, ) @@ -1505,7 +1515,7 @@ func TestApiMetrics(t *testing.T) { defer os.RemoveAll(pluginDir) defer os.RemoveAll(webappPluginDir) - env, err := plugin.NewEnvironment(th.NewPluginAPI, pluginDir, webappPluginDir, th.App.Log(), metricsMock) + env, err := plugin.NewEnvironment(th.NewPluginAPI, NewDriverImpl(th.Server), pluginDir, webappPluginDir, th.App.Log(), metricsMock) require.NoError(t, err) th.App.SetPluginsEnvironment(env) diff --git a/app/plugin_api_tests/test_db_driver/main.go b/app/plugin_api_tests/test_db_driver/main.go new file mode 100644 index 0000000000..f7b5e771b1 --- /dev/null +++ b/app/plugin_api_tests/test_db_driver/main.go @@ -0,0 +1,57 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package main_test + +import ( + "database/sql" + "testing" + + "github.com/mattermost/mattermost-server/v5/app/plugin_api_tests" + "github.com/mattermost/mattermost-server/v5/model" + "github.com/mattermost/mattermost-server/v5/plugin" + "github.com/mattermost/mattermost-server/v5/shared/driver" + "github.com/mattermost/mattermost-server/v5/store/sqlstore" + "github.com/mattermost/mattermost-server/v5/store/storetest" +) + +type MyPlugin struct { + plugin.MattermostPlugin + config plugin_api_tests.BasicConfig + t *testing.T +} + +func (p *MyPlugin) OnConfigurationChange() error { + if err := p.API.LoadPluginConfiguration(&p.config); err != nil { + return err + } + return nil +} + +func (p *MyPlugin) MessageWillBePosted(_ *plugin.Context, _ *model.Post) (*model.Post, string) { + store := sqlstore.New(p.API.GetUnsanitizedConfig().SqlSettings, nil) + store.GetMaster().Db.Close() + + store.GetMaster().Db = sql.OpenDB(driver.NewConnector(p.Driver)) + defer store.GetMaster().Db.Close() + + // Testing with a handful of stores + storetest.TestPostStore(p.t, store, store) + storetest.TestUserStore(p.t, store, store) + storetest.TestTeamStore(p.t, store) + storetest.TestChannelStore(p.t, store, store) + storetest.TestBotStore(p.t, store, store) + + // Use the API to instantiate the driver + // And then run the full suite of tests. + return nil, "OK" +} + +// TestDBAPI is a test function which actually runs a plugin. The objective +// is to run the storetest suite from inside a plugin. +// +// The test runner compiles the test code to a binary, and runs it as a normal +// binary. But under the hood, a test runs. +func TestDBAPI(t *testing.T) { + plugin.ClientMain(&MyPlugin{t: t}) +} diff --git a/app/plugin_db_driver.go b/app/plugin_db_driver.go new file mode 100644 index 0000000000..65e352e926 --- /dev/null +++ b/app/plugin_db_driver.go @@ -0,0 +1,303 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package app + +import ( + "context" + "database/sql" + "database/sql/driver" + "sync" + + "github.com/mattermost/mattermost-server/v5/model" + "github.com/mattermost/mattermost-server/v5/plugin" +) + +// DriverImpl implements the plugin.Driver interface on the server-side. +// Each new request for a connection/statement/transaction etc, generates +// a new entry tracked centrally in a map. Further requests operate on the +// object ID. +type DriverImpl struct { + s *Server + connMut sync.RWMutex + connMap map[string]*sql.Conn + txMut sync.Mutex + txMap map[string]driver.Tx + stMut sync.RWMutex + stMap map[string]driver.Stmt + rowsMut sync.RWMutex + rowsMap map[string]driver.Rows +} + +func NewDriverImpl(s *Server) *DriverImpl { + return &DriverImpl{ + s: s, + connMap: make(map[string]*sql.Conn), + txMap: make(map[string]driver.Tx), + stMap: make(map[string]driver.Stmt), + rowsMap: make(map[string]driver.Rows), + } +} + +func (d *DriverImpl) Conn() (string, error) { + conn, err := d.s.sqlStore.GetMaster().Db.Conn(context.Background()) + if err != nil { + return "", err + } + connID := model.NewId() + d.connMut.Lock() + d.connMap[connID] = conn + d.connMut.Unlock() + return connID, nil +} + +// According to https://golang.org/pkg/database/sql/#Conn, a client can call +// Close on a connection, concurrently while running a query. +// +// Therefore, we have to handle the case where the connection is no longer +// present in the map because it has been closed. ErrBadConn is a good choice +// here which indicates the sql package to retry on a new connection. +// +// ConnPing, ConnQuery, ConnClose, Tx, and Stmt do this. + +func (d *DriverImpl) ConnPing(connID string) error { + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return driver.ErrBadConn + } + + return conn.Raw(func(innerConn interface{}) error { + return innerConn.(driver.Pinger).Ping(context.Background()) + }) +} + +func (d *DriverImpl) ConnQuery(connID, q string, args []driver.NamedValue) (_ string, err error) { + var rows driver.Rows + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return "", driver.ErrBadConn + } + + err = conn.Raw(func(innerConn interface{}) error { + rows, err = innerConn.(driver.QueryerContext).QueryContext(context.Background(), q, args) + return err + }) + if err != nil { + return "", err + } + + rowsID := model.NewId() + d.rowsMut.Lock() + d.rowsMap[rowsID] = rows + d.rowsMut.Unlock() + + return rowsID, nil +} + +func (d *DriverImpl) ConnExec(connID, q string, args []driver.NamedValue) (_ plugin.ResultContainer, err error) { + var res driver.Result + var ret plugin.ResultContainer + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return ret, driver.ErrBadConn + } + + err = conn.Raw(func(innerConn interface{}) error { + res, err = innerConn.(driver.ExecerContext).ExecContext(context.Background(), q, args) + return err + }) + if err != nil { + return ret, err + } + + ret.LastID, ret.LastIDError = res.LastInsertId() + ret.RowsAffected, ret.RowsAffectedError = res.RowsAffected() + + return ret, nil +} + +func (d *DriverImpl) ConnClose(connID string) error { + d.connMut.Lock() + conn, ok := d.connMap[connID] + if !ok { + d.connMut.Unlock() + return driver.ErrBadConn + } + delete(d.connMap, connID) + d.connMut.Unlock() + + return conn.Close() +} + +func (d *DriverImpl) Tx(connID string, opts driver.TxOptions) (_ string, err error) { + var tx driver.Tx + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return "", driver.ErrBadConn + } + + err = conn.Raw(func(innerConn interface{}) error { + tx, err = innerConn.(driver.ConnBeginTx).BeginTx(context.Background(), opts) + return err + }) + if err != nil { + return "", err + } + + txID := model.NewId() + d.txMut.Lock() + d.txMap[txID] = tx + d.txMut.Unlock() + return txID, nil +} + +func (d *DriverImpl) TxCommit(txID string) error { + d.txMut.Lock() + tx := d.txMap[txID] + delete(d.txMap, txID) + d.txMut.Unlock() + + return tx.Commit() +} + +func (d *DriverImpl) TxRollback(txID string) error { + d.txMut.Lock() + tx := d.txMap[txID] + delete(d.txMap, txID) + d.txMut.Unlock() + + return tx.Rollback() +} + +func (d *DriverImpl) Stmt(connID, q string) (_ string, err error) { + var stmt driver.Stmt + d.connMut.RLock() + conn, ok := d.connMap[connID] + d.connMut.RUnlock() + if !ok { + return "", driver.ErrBadConn + } + + err = conn.Raw(func(innerConn interface{}) error { + stmt, err = innerConn.(driver.Conn).Prepare(q) + return err + }) + if err != nil { + return "", err + } + + stID := model.NewId() + d.stMut.Lock() + d.stMap[stID] = stmt + d.stMut.Unlock() + return stID, nil +} + +func (d *DriverImpl) StmtClose(stID string) error { + d.stMut.Lock() + err := d.stMap[stID].Close() + delete(d.stMap, stID) + d.stMut.Unlock() + + return err +} + +func (d *DriverImpl) StmtNumInput(stID string) int { + d.stMut.RLock() + defer d.stMut.RUnlock() + return d.stMap[stID].NumInput() +} + +func (d *DriverImpl) StmtQuery(stID string, args []driver.NamedValue) (string, error) { + argVals := make([]driver.Value, len(args)) + for i, a := range args { + argVals[i] = a.Value + } + d.stMut.RLock() + st := d.stMap[stID] + d.stMut.RUnlock() + + rows, err := st.Query(argVals) //nolint:staticcheck + if err != nil { + return "", err + } + rowsID := model.NewId() + d.rowsMut.Lock() + d.rowsMap[rowsID] = rows + d.rowsMut.Unlock() + return rowsID, nil +} + +func (d *DriverImpl) StmtExec(stID string, args []driver.NamedValue) (plugin.ResultContainer, error) { + argVals := make([]driver.Value, len(args)) + for i, a := range args { + argVals[i] = a.Value + } + var ret plugin.ResultContainer + d.stMut.RLock() + st := d.stMap[stID] + d.stMut.RUnlock() + + res, err := st.Exec(argVals) //nolint:staticcheck + if err != nil { + return ret, err + } + + ret.LastID, ret.LastIDError = res.LastInsertId() + ret.RowsAffected, ret.RowsAffectedError = res.RowsAffected() + + return ret, nil +} + +func (d *DriverImpl) RowsColumns(rowsID string) []string { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].Columns() +} + +func (d *DriverImpl) RowsClose(rowsID string) error { + d.rowsMut.Lock() + defer d.rowsMut.Unlock() + err := d.rowsMap[rowsID].Close() + delete(d.rowsMap, rowsID) + return err +} + +func (d *DriverImpl) RowsNext(rowsID string, dest []driver.Value) error { + d.rowsMut.RLock() + rows := d.rowsMap[rowsID] + d.rowsMut.RUnlock() + return rows.Next(dest) +} + +func (d *DriverImpl) RowsHasNextResultSet(rowsID string) bool { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].(driver.RowsNextResultSet).HasNextResultSet() +} + +func (d *DriverImpl) RowsNextResultSet(rowsID string) error { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].(driver.RowsNextResultSet).NextResultSet() +} + +func (d *DriverImpl) RowsColumnTypeDatabaseTypeName(rowsID string, index int) string { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].(driver.RowsColumnTypeDatabaseTypeName).ColumnTypeDatabaseTypeName(index) +} + +func (d *DriverImpl) RowsColumnTypePrecisionScale(rowsID string, index int) (int64, int64, bool) { + d.rowsMut.RLock() + defer d.rowsMut.RUnlock() + return d.rowsMap[rowsID].(driver.RowsColumnTypePrecisionScale).ColumnTypePrecisionScale(index) +} diff --git a/app/plugin_hooks_test.go b/app/plugin_hooks_test.go index 02edfbfdb8..e429ea1756 100644 --- a/app/plugin_hooks_test.go +++ b/app/plugin_hooks_test.go @@ -34,7 +34,7 @@ func SetAppEnvironmentWithPlugins(t *testing.T, pluginCode []string, app *App, a webappPluginDir, err := ioutil.TempDir("", "") require.NoError(t, err) - env, err := plugin.NewEnvironment(apiFunc, pluginDir, webappPluginDir, app.Log(), nil) + env, err := plugin.NewEnvironment(apiFunc, NewDriverImpl(app.srv), pluginDir, webappPluginDir, app.Log(), nil) require.NoError(t, err) app.SetPluginsEnvironment(env) @@ -1045,7 +1045,7 @@ func TestHookMetrics(t *testing.T) { defer os.RemoveAll(pluginDir) defer os.RemoveAll(webappPluginDir) - env, err := plugin.NewEnvironment(th.NewPluginAPI, pluginDir, webappPluginDir, th.App.Log(), metricsMock) + env, err := plugin.NewEnvironment(th.NewPluginAPI, NewDriverImpl(th.Server), pluginDir, webappPluginDir, th.App.Log(), metricsMock) require.NoError(t, err) th.App.SetPluginsEnvironment(env) diff --git a/plugin/client.go b/plugin/client.go index ed3d872d79..15ac0254c9 100644 --- a/plugin/client.go +++ b/plugin/client.go @@ -19,11 +19,13 @@ func ClientMain(pluginImplementation interface{}) { if impl, ok := pluginImplementation.(interface { SetAPI(api API) SetHelpers(helpers Helpers) + SetDriver(driver Driver) }); !ok { panic("Plugin implementation given must embed plugin.MattermostPlugin") } else { impl.SetAPI(nil) impl.SetHelpers(nil) + impl.SetDriver(nil) } pluginMap := map[string]plugin.Plugin{ @@ -40,6 +42,7 @@ type MattermostPlugin struct { // API exposes the plugin api, and becomes available just prior to the OnActive hook. API API Helpers Helpers + Driver Driver } // SetAPI persists the given API interface to the plugin. It is invoked just prior to the @@ -52,3 +55,7 @@ func (p *MattermostPlugin) SetAPI(api API) { func (p *MattermostPlugin) SetHelpers(helpers Helpers) { p.Helpers = helpers } + +func (p *MattermostPlugin) SetDriver(driver Driver) { + p.Driver = driver +} diff --git a/plugin/client_rpc.go b/plugin/client_rpc.go index 8382fe005e..58c0841a8c 100644 --- a/plugin/client_rpc.go +++ b/plugin/client_rpc.go @@ -7,6 +7,8 @@ package plugin import ( "bytes" + "database/sql" + "database/sql/driver" "encoding/gob" "encoding/json" "fmt" @@ -19,7 +21,9 @@ import ( "reflect" "github.com/dyatlov/go-opengraph/opengraph" + "github.com/go-sql-driver/mysql" "github.com/hashicorp/go-plugin" + "github.com/lib/pq" "github.com/mattermost/mattermost-server/v5/model" "github.com/mattermost/mattermost-server/v5/shared/mlog" @@ -32,6 +36,7 @@ type hooksRPCClient struct { log *mlog.Logger muxBroker *plugin.MuxBroker apiImpl API + driver Driver implemented [TotalHooksID]bool } @@ -43,9 +48,10 @@ type hooksRPCServer struct { // Implements hashicorp/go-plugin/plugin.Plugin interface to connect the hooks of a plugin type hooksPlugin struct { - hooks interface{} - apiImpl API - log *mlog.Logger + hooks interface{} + apiImpl API + driverImpl Driver + log *mlog.Logger } func (p *hooksPlugin) Server(b *plugin.MuxBroker) (interface{}, error) { @@ -53,7 +59,12 @@ func (p *hooksPlugin) Server(b *plugin.MuxBroker) (interface{}, error) { } func (p *hooksPlugin) Client(b *plugin.MuxBroker, client *rpc.Client) (interface{}, error) { - return &hooksRPCClient{client: client, log: p.log, muxBroker: b, apiImpl: p.apiImpl}, nil + return &hooksRPCClient{client: client, + log: p.log, + muxBroker: b, + apiImpl: p.apiImpl, + driver: p.driverImpl, + }, nil } type apiRPCClient struct { @@ -72,7 +83,8 @@ type apiRPCServer struct { // ErrorString merely preserves the string description of the error, while satisfying the error // interface itself to allow other registered types (such as model.AppError) to be sent unmodified. type ErrorString struct { - Err string + Code int // Code to map to various error variables + Err string } func (e ErrorString) Error() string { @@ -87,9 +99,58 @@ func encodableError(err error) error { return err } - return &ErrorString{ + if _, ok := err.(*pq.Error); ok { + return err + } + + if _, ok := err.(*mysql.MySQLError); ok { + return err + } + + ret := &ErrorString{ Err: err.Error(), } + + switch err { + case io.EOF: + ret.Code = 1 + case sql.ErrNoRows: + ret.Code = 2 + case sql.ErrConnDone: + ret.Code = 3 + case sql.ErrTxDone: + ret.Code = 4 + case driver.ErrSkip: + ret.Code = 5 + case driver.ErrBadConn: + ret.Code = 6 + case driver.ErrRemoveArgument: + ret.Code = 7 + } + + return ret +} + +func decodableError(err error) error { + if encErr, ok := err.(*ErrorString); ok { + switch encErr.Code { + case 1: + return io.EOF + case 2: + return sql.ErrNoRows + case 3: + return sql.ErrConnDone + case 4: + return sql.ErrTxDone + case 5: + return driver.ErrSkip + case 6: + return driver.ErrBadConn + case 7: + return driver.ErrRemoveArgument + } + } + return err } // Registering some types used by MM for encoding/gob used by rpc @@ -98,6 +159,8 @@ func init() { gob.Register([]interface{}{}) gob.Register(map[string]interface{}{}) gob.Register(&model.AppError{}) + gob.Register(&pq.Error{}) + gob.Register(&mysql.MySQLError{}) gob.Register(&ErrorString{}) gob.Register(&opengraph.OpenGraph{}) gob.Register(&model.AutocompleteDynamicListArg{}) @@ -167,7 +230,8 @@ func (s *hooksRPCServer) Implemented(args struct{}, reply *[]string) error { } type Z_OnActivateArgs struct { - APIMuxId uint32 + APIMuxId uint32 + DriverMuxId uint32 } type Z_OnActivateReturns struct { @@ -181,8 +245,14 @@ func (g *hooksRPCClient) OnActivate() error { muxBroker: g.muxBroker, }) + nextID := g.muxBroker.NextId() + go g.muxBroker.AcceptAndServe(nextID, &dbRPCServer{ + dbImpl: g.driver, + }) + _args := &Z_OnActivateArgs{ - APIMuxId: muxId, + APIMuxId: muxId, + DriverMuxId: nextID, } _returns := &Z_OnActivateReturns{} @@ -198,17 +268,28 @@ func (s *hooksRPCServer) OnActivate(args *Z_OnActivateArgs, returns *Z_OnActivat return err } + conn2, err := s.muxBroker.Dial(args.DriverMuxId) + if err != nil { + return err + } + s.apiRPCClient = &apiRPCClient{ client: rpc.NewClient(connection), muxBroker: s.muxBroker, } + dbClient := &dbRPCClient{ + client: rpc.NewClient(conn2), + } + if mmplugin, ok := s.impl.(interface { SetAPI(api API) SetHelpers(helpers Helpers) + SetDriver(driver Driver) }); ok { mmplugin.SetAPI(s.apiRPCClient) mmplugin.SetHelpers(&HelpersImpl{API: s.apiRPCClient}) + mmplugin.SetDriver(dbClient) } if mmplugin, ok := s.impl.(interface { diff --git a/plugin/db_rpc.go b/plugin/db_rpc.go new file mode 100644 index 0000000000..0d2701f24d --- /dev/null +++ b/plugin/db_rpc.go @@ -0,0 +1,461 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package plugin + +import ( + "database/sql/driver" + "log" + "net/rpc" +) + +// dbRPCClient contains the client-side logic to handle the RPC communication +// with the server. It's API is hand-written because we do not expect +// new methods to be added very frequently. +type dbRPCClient struct { + client *rpc.Client +} + +// dbRPCServer is the server-side component which is responsible for calling +// the driver methods and properly encoding the responses back to the RPC client. +type dbRPCServer struct { + dbImpl Driver +} + +var _ Driver = &dbRPCClient{} + +type Z_DbStrErrReturn struct { + A string + B error +} + +type Z_DbErrReturn struct { + A error +} + +type Z_DbInt64ErrReturn struct { + A int64 + B error +} + +type Z_DbBoolReturn struct { + A bool +} + +func (db *dbRPCClient) Conn() (string, error) { + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.Conn", struct{}{}, ret) + if err != nil { + log.Printf("error during Plugin.Conn: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) Conn(_ struct{}, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.Conn() + ret.B = encodableError(ret.B) + return nil +} + +func (db *dbRPCClient) ConnPing(connID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.ConnPing", connID, ret) + if err != nil { + log.Printf("error during Plugin.ConnPing: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) ConnPing(connID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.ConnPing(connID) + ret.A = encodableError(ret.A) + return nil +} + +func (db *dbRPCClient) ConnClose(connID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.ConnClose", connID, ret) + if err != nil { + log.Printf("error during Plugin.ConnClose: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) ConnClose(connID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.ConnClose(connID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbTxArgs struct { + A string + B driver.TxOptions +} + +func (db *dbRPCClient) Tx(connID string, opts driver.TxOptions) (string, error) { + args := &Z_DbTxArgs{ + A: connID, + B: opts, + } + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.Tx", args, ret) + if err != nil { + log.Printf("error during Plugin.Tx: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) Tx(args *Z_DbTxArgs, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.Tx(args.A, args.B) + ret.B = encodableError(ret.B) + return nil +} + +func (db *dbRPCClient) TxCommit(txID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.TxCommit", txID, ret) + if err != nil { + log.Printf("error during Plugin.TxCommit: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) TxCommit(txID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.TxCommit(txID) + ret.A = encodableError(ret.A) + return nil +} + +func (db *dbRPCClient) TxRollback(txID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.TxRollback", txID, ret) + if err != nil { + log.Printf("error during Plugin.TxRollback: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) TxRollback(txID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.TxRollback(txID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbStmtArgs struct { + A string + B string +} + +func (db *dbRPCClient) Stmt(connID, q string) (string, error) { + args := &Z_DbStmtArgs{ + A: connID, + B: q, + } + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.Stmt", args, ret) + if err != nil { + log.Printf("error during Plugin.Stmt: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) Stmt(args *Z_DbStmtArgs, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.Stmt(args.A, args.B) + ret.B = encodableError(ret.B) + return nil +} + +func (db *dbRPCClient) StmtClose(stID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.StmtClose", stID, ret) + if err != nil { + log.Printf("error during Plugin.StmtClose: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) StmtClose(stID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.StmtClose(stID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbIntReturn struct { + A int +} + +func (db *dbRPCClient) StmtNumInput(stID string) int { + ret := &Z_DbIntReturn{} + err := db.client.Call("Plugin.StmtNumInput", stID, ret) + if err != nil { + log.Printf("error during Plugin.StmtNumInput: %v", err) + } + return ret.A +} + +func (db *dbRPCServer) StmtNumInput(stID string, ret *Z_DbIntReturn) error { + ret.A = db.dbImpl.StmtNumInput(stID) + return nil +} + +type Z_DbStmtQueryArgs struct { + A string + B []driver.NamedValue +} + +func (db *dbRPCClient) StmtQuery(stID string, argVals []driver.NamedValue) (string, error) { + args := &Z_DbStmtQueryArgs{ + A: stID, + B: argVals, + } + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.StmtQuery", args, ret) + if err != nil { + log.Printf("error during Plugin.StmtQuery: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) StmtQuery(args *Z_DbStmtQueryArgs, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.StmtQuery(args.A, args.B) + ret.B = encodableError(ret.B) + return nil +} + +func (db *dbRPCClient) StmtExec(stID string, argVals []driver.NamedValue) (ResultContainer, error) { + args := &Z_DbStmtQueryArgs{ + A: stID, + B: argVals, + } + ret := &Z_DbResultContErrReturn{} + err := db.client.Call("Plugin.StmtExec", args, ret) + if err != nil { + log.Printf("error during Plugin.StmtExec: %v", err) + } + ret.A.LastIDError = decodableError(ret.A.LastIDError) + ret.A.RowsAffectedError = decodableError(ret.A.RowsAffectedError) + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) StmtExec(args *Z_DbStmtQueryArgs, ret *Z_DbResultContErrReturn) error { + ret.A, ret.B = db.dbImpl.StmtExec(args.A, args.B) + ret.A.LastIDError = encodableError(ret.A.LastIDError) + ret.A.RowsAffectedError = encodableError(ret.A.RowsAffectedError) + ret.B = encodableError(ret.B) + return nil +} + +type Z_DbConnArgs struct { + A string + B string + C []driver.NamedValue +} + +func (db *dbRPCClient) ConnQuery(connID, q string, argVals []driver.NamedValue) (string, error) { + args := &Z_DbConnArgs{ + A: connID, + B: q, + C: argVals, + } + ret := &Z_DbStrErrReturn{} + err := db.client.Call("Plugin.ConnQuery", args, ret) + if err != nil { + log.Printf("error during Plugin.ConnQuery: %v", err) + } + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) ConnQuery(args *Z_DbConnArgs, ret *Z_DbStrErrReturn) error { + ret.A, ret.B = db.dbImpl.ConnQuery(args.A, args.B, args.C) + ret.B = encodableError(ret.B) + return nil +} + +type Z_DbResultContErrReturn struct { + A ResultContainer + B error +} + +func (db *dbRPCClient) ConnExec(connID, q string, argVals []driver.NamedValue) (ResultContainer, error) { + args := &Z_DbConnArgs{ + A: connID, + B: q, + C: argVals, + } + ret := &Z_DbResultContErrReturn{} + err := db.client.Call("Plugin.ConnExec", args, ret) + if err != nil { + log.Printf("error during Plugin.ConnExec: %v", err) + } + ret.A.LastIDError = decodableError(ret.A.LastIDError) + ret.A.RowsAffectedError = decodableError(ret.A.RowsAffectedError) + ret.B = decodableError(ret.B) + return ret.A, ret.B +} + +func (db *dbRPCServer) ConnExec(args *Z_DbConnArgs, ret *Z_DbResultContErrReturn) error { + ret.A, ret.B = db.dbImpl.ConnExec(args.A, args.B, args.C) + ret.A.LastIDError = encodableError(ret.A.LastIDError) + ret.A.RowsAffectedError = encodableError(ret.A.RowsAffectedError) + ret.B = encodableError(ret.B) + return nil +} + +type Z_DbStrSliceReturn struct { + A []string +} + +func (db *dbRPCClient) RowsColumns(rowsID string) []string { + ret := &Z_DbStrSliceReturn{} + err := db.client.Call("Plugin.RowsColumns", rowsID, ret) + if err != nil { + log.Printf("error during Plugin.RowsColumns: %v", err) + } + return ret.A +} + +func (db *dbRPCServer) RowsColumns(rowsID string, ret *Z_DbStrSliceReturn) error { + ret.A = db.dbImpl.RowsColumns(rowsID) + return nil +} + +func (db *dbRPCClient) RowsClose(resID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.RowsClose", resID, ret) + if err != nil { + log.Printf("error during Plugin.RowsClose: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) RowsClose(resID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.RowsClose(resID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbRowScanReturn struct { + A error + B []driver.Value +} + +type Z_DbRowScanArg struct { + A string + B []driver.Value +} + +func (db *dbRPCClient) RowsNext(rowsID string, dest []driver.Value) error { + args := &Z_DbRowScanArg{ + A: rowsID, + B: dest, + } + ret := &Z_DbRowScanReturn{} + err := db.client.Call("Plugin.RowsNext", args, ret) + if err != nil { + log.Printf("error during Plugin.RowsNext: %v", err) + } + ret.A = decodableError(ret.A) + for i, v := range ret.B { + dest[i] = v + } + return ret.A +} + +func (db *dbRPCServer) RowsNext(args *Z_DbRowScanArg, ret *Z_DbRowScanReturn) error { + ret.A = db.dbImpl.RowsNext(args.A, args.B) + ret.A = encodableError(ret.A) + // Trick to populate the dest slice. RPC doesn't have a semantic to populate + // pointer type args. So the only way to pass values is via args, and only way + // to return values is via the return struct. + ret.B = args.B + return nil +} + +func (db *dbRPCClient) RowsHasNextResultSet(rowsID string) bool { + ret := &Z_DbBoolReturn{} + err := db.client.Call("Plugin.RowsHasNextResultSet", rowsID, ret) + if err != nil { + log.Printf("error during Plugin.RowsHasNextResultSet: %v", err) + } + return ret.A +} + +func (db *dbRPCServer) RowsHasNextResultSet(rowsID string, ret *Z_DbBoolReturn) error { + ret.A = db.dbImpl.RowsHasNextResultSet(rowsID) + return nil +} + +func (db *dbRPCClient) RowsNextResultSet(rowsID string) error { + ret := &Z_DbErrReturn{} + err := db.client.Call("Plugin.RowsNextResultSet", rowsID, ret) + if err != nil { + log.Printf("error during Plugin.RowsNextResultSet: %v", err) + } + ret.A = decodableError(ret.A) + return ret.A +} + +func (db *dbRPCServer) RowsNextResultSet(rowsID string, ret *Z_DbErrReturn) error { + ret.A = db.dbImpl.RowsNextResultSet(rowsID) + ret.A = encodableError(ret.A) + return nil +} + +type Z_DbRowsColumnArg struct { + A string + B int +} + +func (db *dbRPCClient) RowsColumnTypeDatabaseTypeName(rowsID string, index int) string { + args := &Z_DbRowsColumnArg{ + A: rowsID, + B: index, + } + var ret string + err := db.client.Call("Plugin.RowsColumnTypeDatabaseTypeName", args, &ret) + if err != nil { + log.Printf("error during Plugin.RowsColumnTypeDatabaseTypeName: %v", err) + } + return ret +} + +func (db *dbRPCServer) RowsColumnTypeDatabaseTypeName(args *Z_DbRowsColumnArg, ret *string) error { + *ret = db.dbImpl.RowsColumnTypeDatabaseTypeName(args.A, args.B) + return nil +} + +type Z_DbRowsColumnTypePrecisionScaleReturn struct { + A int64 + B int64 + C bool +} + +func (db *dbRPCClient) RowsColumnTypePrecisionScale(rowsID string, index int) (int64, int64, bool) { + args := &Z_DbRowsColumnArg{ + A: rowsID, + B: index, + } + ret := &Z_DbRowsColumnTypePrecisionScaleReturn{} + err := db.client.Call("Plugin.RowsColumnTypePrecisionScale", args, ret) + if err != nil { + log.Printf("error during Plugin.RowsColumnTypePrecisionScale: %v", err) + } + return ret.A, ret.B, ret.C +} + +func (db *dbRPCServer) RowsColumnTypePrecisionScale(args *Z_DbRowsColumnArg, ret *Z_DbRowsColumnTypePrecisionScaleReturn) error { + ret.A, ret.B, ret.C = db.dbImpl.RowsColumnTypePrecisionScale(args.A, args.B) + return nil +} diff --git a/plugin/environment.go b/plugin/environment.go index 6d44047dde..50c2b037cd 100644 --- a/plugin/environment.go +++ b/plugin/environment.go @@ -54,17 +54,23 @@ type Environment struct { logger *mlog.Logger metrics einterfaces.MetricsInterface newAPIImpl apiImplCreatorFunc + dbDriver Driver pluginDir string webappPluginDir string prepackagedPlugins []*PrepackagedPlugin prepackagedPluginsLock sync.RWMutex } -func NewEnvironment(newAPIImpl apiImplCreatorFunc, pluginDir string, webappPluginDir string, logger *mlog.Logger, metrics einterfaces.MetricsInterface) (*Environment, error) { +func NewEnvironment(newAPIImpl apiImplCreatorFunc, + dbDriver Driver, + pluginDir string, webappPluginDir string, + logger *mlog.Logger, + metrics einterfaces.MetricsInterface) (*Environment, error) { return &Environment{ logger: logger, metrics: metrics, newAPIImpl: newAPIImpl, + dbDriver: dbDriver, pluginDir: pluginDir, webappPluginDir: webappPluginDir, }, nil @@ -263,7 +269,7 @@ func (env *Environment) Activate(id string) (manifest *model.Manifest, activated } if pluginInfo.Manifest.HasServer() { - sup, err := newSupervisor(pluginInfo, env.newAPIImpl(pluginInfo.Manifest), env.logger, env.metrics) + sup, err := newSupervisor(pluginInfo, env.newAPIImpl(pluginInfo.Manifest), env.dbDriver, env.logger, env.metrics) if err != nil { return nil, false, errors.Wrapf(err, "unable to start plugin: %v", id) } diff --git a/plugin/health_check_test.go b/plugin/health_check_test.go index 7aa559576c..8a7d929a26 100644 --- a/plugin/health_check_test.go +++ b/plugin/health_check_test.go @@ -59,7 +59,7 @@ func testPluginHealthCheckSuccess(t *testing.T) { EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) require.NoError(t, err) require.NotNil(t, supervisor) defer supervisor.Shutdown() @@ -106,7 +106,7 @@ func testPluginHealthCheckPanic(t *testing.T) { EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) require.NoError(t, err) require.NotNil(t, supervisor) defer supervisor.Shutdown() diff --git a/plugin/helpers.go b/plugin/helpers.go index 914fc2e1f2..68885dd78f 100644 --- a/plugin/helpers.go +++ b/plugin/helpers.go @@ -4,6 +4,8 @@ package plugin import ( + "database/sql/driver" + "github.com/mattermost/mattermost-server/v5/model" ) @@ -95,3 +97,55 @@ type Helpers interface { type HelpersImpl struct { API API } + +// ResultContainer contains the output from the LastInsertID +// and RowsAffected methods for a given set of rows. +// It is used to embed another round-trip to the server, +// and helping to avoid tracking results on the server. +type ResultContainer struct { + LastID int64 + LastIDError error + RowsAffected int64 + RowsAffectedError error +} + +type Driver interface { + // Connection + Conn() (string, error) + ConnPing(connID string) error + ConnClose(connID string) error + ConnQuery(connID, q string, args []driver.NamedValue) (string, error) // rows + ConnExec(connID, q string, args []driver.NamedValue) (ResultContainer, error) // result + + // Transaction + Tx(connID string, opts driver.TxOptions) (string, error) + TxCommit(txID string) error + TxRollback(txID string) error + + // Statement + Stmt(connID, q string) (string, error) + StmtClose(stID string) error + StmtNumInput(stID string) int + StmtQuery(stID string, args []driver.NamedValue) (string, error) // rows + StmtExec(stID string, args []driver.NamedValue) (ResultContainer, error) // result + + // Rows + RowsColumns(rowsID string) []string + RowsClose(rowsID string) error + RowsNext(rowsID string, dest []driver.Value) error + RowsHasNextResultSet(rowsID string) bool + RowsNextResultSet(rowsID string) error + RowsColumnTypeDatabaseTypeName(rowsID string, index int) string + RowsColumnTypePrecisionScale(rowsID string, index int) (int64, int64, bool) + + // TODO: add this + // RowsColumnScanType(rowsID string, index int) reflect.Type + + // Note: the following cannot be implemented because either MySQL or PG + // does not support it. So this implementation has to be a common subset + // of both DB implementations. + // RowsColumnTypeLength(rowsID string, index int) (int64, bool) + // RowsColumnTypeNullable(rowsID string, index int) (bool, bool) + // ResetSession(ctx context.Context) error + // IsValid() bool +} diff --git a/plugin/supervisor.go b/plugin/supervisor.go index ce067aff39..de1fbf3715 100644 --- a/plugin/supervisor.go +++ b/plugin/supervisor.go @@ -27,7 +27,7 @@ type supervisor struct { pid int } -func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, parentLogger *mlog.Logger, metrics einterfaces.MetricsInterface) (retSupervisor *supervisor, retErr error) { +func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, driver Driver, parentLogger *mlog.Logger, metrics einterfaces.MetricsInterface) (retSupervisor *supervisor, retErr error) { sup := supervisor{} defer func() { if retErr != nil { @@ -44,8 +44,9 @@ func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, parentLogger *mlog pluginMap := map[string]plugin.Plugin{ "hooks": &hooksPlugin{ - log: wrappedLogger, - apiImpl: &apiTimerLayer{pluginInfo.Manifest.Id, apiImpl, metrics}, + log: wrappedLogger, + driverImpl: driver, + apiImpl: &apiTimerLayer{pluginInfo.Manifest.Id, apiImpl, metrics}, }, } diff --git a/plugin/supervisor_test.go b/plugin/supervisor_test.go index 0f2bdef9fa..921841b704 100644 --- a/plugin/supervisor_test.go +++ b/plugin/supervisor_test.go @@ -41,7 +41,7 @@ func testSupervisorInvalidExecutablePath(t *testing.T) { ConsoleLevel: "error", EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) assert.Nil(t, supervisor) assert.Error(t, err) } @@ -60,7 +60,7 @@ func testSupervisorNonExistentExecutablePath(t *testing.T) { ConsoleLevel: "error", EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) require.Error(t, err) require.Nil(t, supervisor) } @@ -90,7 +90,7 @@ func testSupervisorStartTimeout(t *testing.T) { ConsoleLevel: "error", EnableFile: false, }) - supervisor, err := newSupervisor(bundle, nil, log, nil) + supervisor, err := newSupervisor(bundle, nil, nil, log, nil) require.Error(t, err) require.Nil(t, supervisor) } diff --git a/services/telemetry/telemetry_test.go b/services/telemetry/telemetry_test.go index 207ec8d5e1..9a57ee8320 100644 --- a/services/telemetry/telemetry_test.go +++ b/services/telemetry/telemetry_test.go @@ -52,7 +52,12 @@ func initializeMocks(cfg *model.Config) (*mocks.ServerIface, *storeMocks.Store, os.RemoveAll(webappPluginDir) } pluginsAPIMock := &plugintest.API{} - pluginEnv, _ := plugin.NewEnvironment(func(m *model.Manifest) plugin.API { return pluginsAPIMock }, pluginDir, webappPluginDir, mlog.NewLogger(&mlog.LoggerConfiguration{}), nil) + pluginEnv, _ := plugin.NewEnvironment( + func(m *model.Manifest) plugin.API { return pluginsAPIMock }, + nil, + pluginDir, webappPluginDir, + mlog.NewLogger(&mlog.LoggerConfiguration{}), + nil) serverIfaceMock.On("GetPluginsEnvironment").Return(pluginEnv, nil) serverIfaceMock.On("License").Return(model.NewTestLicense(), nil) diff --git a/shared/driver/conn.go b/shared/driver/conn.go index 196a1ad892..c3cfa0e548 100644 --- a/shared/driver/conn.go +++ b/shared/driver/conn.go @@ -5,15 +5,16 @@ package driver import ( "context" - "database/sql" "database/sql/driver" + + "github.com/mattermost/mattermost-server/v5/plugin" ) // Conn is a DB driver conn implementation -// which will just pass-through all queries to its -// underlying connection. +// which executes queries using the Plugin DB API. type Conn struct { - conn *sql.Conn + id string + api plugin.Driver } // driverConn is a super-interface combining the basic @@ -33,66 +34,84 @@ var ( ) func (c *Conn) Begin() (tx driver.Tx, err error) { - err = c.conn.Raw(func(innerConn interface{}) error { - tx, err = innerConn.(driver.Conn).Begin() //nolint:staticcheck - return err - }) - return tx, err + txID, err := c.api.Tx(c.id, driver.TxOptions{}) + if err != nil { + return nil, err + } + + t := &wrapperTx{ + id: txID, + api: c.api, + } + return t, nil } -func (c *Conn) BeginTx(ctx context.Context, opts driver.TxOptions) (_ driver.Tx, err error) { - t := &wrapperTx{} - err = c.conn.Raw(func(innerConn interface{}) error { - t.Tx, err = innerConn.(driver.ConnBeginTx).BeginTx(ctx, opts) - return err - }) - return t, err +func (c *Conn) BeginTx(_ context.Context, opts driver.TxOptions) (driver.Tx, error) { + txID, err := c.api.Tx(c.id, opts) + if err != nil { + return nil, err + } + + t := &wrapperTx{ + id: txID, + api: c.api, + } + return t, nil } -func (c *Conn) Prepare(q string) (_ driver.Stmt, err error) { - st := &wrapperStmt{} - err = c.conn.Raw(func(innerConn interface{}) error { - st.Stmt, err = innerConn.(driver.Conn).Prepare(q) - return err - }) - return st, err +func (c *Conn) Prepare(q string) (driver.Stmt, error) { + stID, err := c.api.Stmt(c.id, q) + if err != nil { + return nil, err + } + + st := &wrapperStmt{ + id: stID, + api: c.api, + } + return st, nil } -func (c *Conn) PrepareContext(ctx context.Context, q string) (_ driver.Stmt, err error) { - st := &wrapperStmt{} - err = c.conn.Raw(func(innerConn interface{}) error { - st.Stmt, err = innerConn.(driver.ConnPrepareContext).PrepareContext(ctx, q) - return err - }) - return st, err +func (c *Conn) PrepareContext(_ context.Context, q string) (driver.Stmt, error) { + stID, err := c.api.Stmt(c.id, q) + if err != nil { + return nil, err + } + st := &wrapperStmt{ + id: stID, + api: c.api, + } + return st, nil } -func (c *Conn) ExecContext(ctx context.Context, q string, args []driver.NamedValue) (_ driver.Result, err error) { - res := &wrapperResult{} - err = c.conn.Raw(func(innerConn interface{}) error { - res.Result, err = innerConn.(driver.ExecerContext).ExecContext(ctx, q, args) - return err - }) - return res, err +func (c *Conn) ExecContext(_ context.Context, q string, args []driver.NamedValue) (driver.Result, error) { + resultContainer, err := c.api.ConnExec(c.id, q, args) + if err != nil { + return nil, err + } + res := &wrapperResult{ + res: resultContainer, + } + return res, nil } -func (c *Conn) QueryContext(ctx context.Context, q string, args []driver.NamedValue) (_ driver.Rows, err error) { - rows := &wrapperRows{} - err = c.conn.Raw(func(innerConn interface{}) error { - rows.Rows, err = innerConn.(driver.QueryerContext).QueryContext(ctx, q, args) - return err - }) - return rows, err +func (c *Conn) QueryContext(_ context.Context, q string, args []driver.NamedValue) (driver.Rows, error) { + rowsID, err := c.api.ConnQuery(c.id, q, args) + if err != nil { + return nil, err + } + + rows := &wrapperRows{ + id: rowsID, + api: c.api, + } + return rows, nil } -func (c *Conn) Ping(ctx context.Context) error { - return c.conn.Raw(func(innerConn interface{}) error { - return innerConn.(driver.Pinger).Ping(ctx) - }) +func (c *Conn) Ping(_ context.Context) error { + return c.api.ConnPing(c.id) } func (c *Conn) Close() error { - return c.conn.Raw(func(innerConn interface{}) error { - return innerConn.(driver.Conn).Close() - }) + return c.api.ConnClose(c.id) } diff --git a/shared/driver/driver.go b/shared/driver/driver.go index a38854a4b9..a917c87404 100644 --- a/shared/driver/driver.go +++ b/shared/driver/driver.go @@ -1,12 +1,18 @@ // Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. // See LICENSE.txt for license information. +// package driver implements a DB driver that can be used by plugins +// to make SQL queries using RPC. This helps to avoid opening new connections +// for every plugin, and lets everyone use the central connection +// pool in the server. +// The tests for this package are at app/plugin_api_tests/test_db_driver/main.go. package driver import ( "context" - "database/sql" "database/sql/driver" + + "github.com/mattermost/mattermost-server/v5/plugin" ) var ( @@ -15,32 +21,22 @@ var ( ) // Connector is the DB connector which is used to -// initialize the underlying DB. +// communicate with the DB API. type Connector struct { - driverName string - dsn string - db *sql.DB + api plugin.Driver } -func NewConnector(driverName, dsn string) (*Connector, error) { - db, err := sql.Open(driverName, dsn) - if err != nil { - return nil, err - } - return &Connector{ - driverName: driverName, - dsn: dsn, - db: db, - }, nil +func NewConnector(api plugin.Driver) *Connector { + return &Connector{api: api} } -func (c *Connector) Connect(ctx context.Context) (driver.Conn, error) { - conn, err := c.db.Conn(ctx) +func (c *Connector) Connect(_ context.Context) (driver.Conn, error) { + connID, err := c.api.Conn() if err != nil { return nil, err } - return &Conn{conn: conn}, nil + return &Conn{id: connID, api: c.api}, nil } func (c *Connector) Driver() driver.Driver { diff --git a/shared/driver/objects.go b/shared/driver/objects.go index 008c189ccf..f6c5ef19e5 100644 --- a/shared/driver/objects.go +++ b/shared/driver/objects.go @@ -6,93 +6,109 @@ package driver import ( "context" "database/sql/driver" - "reflect" + + "github.com/mattermost/mattermost-server/v5/plugin" ) type wrapperTx struct { driver.Tx + id string + api plugin.Driver } func (t *wrapperTx) Commit() error { - return t.Tx.Commit() + return t.api.TxCommit(t.id) } func (t *wrapperTx) Rollback() error { - return t.Tx.Rollback() + return t.api.TxRollback(t.id) } type wrapperStmt struct { driver.Stmt + id string + api plugin.Driver } func (s *wrapperStmt) Close() error { - return s.Stmt.Close() + return s.api.StmtClose(s.id) } func (s *wrapperStmt) NumInput() int { - return s.Stmt.NumInput() + return s.api.StmtNumInput(s.id) } -func (s *wrapperStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) { - return s.Stmt.(driver.StmtExecContext).ExecContext(ctx, args) +func (s *wrapperStmt) ExecContext(_ context.Context, args []driver.NamedValue) (driver.Result, error) { + resultContainer, err := s.api.StmtExec(s.id, args) + if err != nil { + return nil, err + } + res := &wrapperResult{ + res: resultContainer, + } + return res, nil } -func (s *wrapperStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) { - return s.Stmt.(driver.StmtQueryContext).QueryContext(ctx, args) +func (s *wrapperStmt) QueryContext(_ context.Context, args []driver.NamedValue) (driver.Rows, error) { + rowsID, err := s.api.StmtQuery(s.id, args) + if err != nil { + return nil, err + } + rows := &wrapperRows{ + id: rowsID, + api: s.api, + } + return rows, nil } +// wrapperResult implements the driver.Result interface. +// This differs from other objects because it already contains the +// information for its methods. This does two things: +// +// 1. Simplifies server-side code by avoiding to track result ids +// in a map. +// 2. Avoids round-trip to compute result methods. type wrapperResult struct { - driver.Result + res plugin.ResultContainer } func (r *wrapperResult) LastInsertId() (int64, error) { - return r.Result.LastInsertId() + return r.res.LastID, r.res.LastIDError } func (r *wrapperResult) RowsAffected() (int64, error) { - return r.Result.RowsAffected() + return r.res.RowsAffected, r.res.RowsAffectedError } type wrapperRows struct { - driver.Rows + id string + api plugin.Driver } func (r *wrapperRows) Columns() []string { - return r.Rows.Columns() + return r.api.RowsColumns(r.id) } func (r *wrapperRows) Close() error { - return r.Rows.Close() + return r.api.RowsClose(r.id) } func (r *wrapperRows) Next(dest []driver.Value) error { - return r.Rows.Next(dest) + return r.api.RowsNext(r.id, dest) } func (r *wrapperRows) HasNextResultSet() bool { - return r.Rows.(driver.RowsNextResultSet).HasNextResultSet() + return r.api.RowsHasNextResultSet(r.id) } func (r *wrapperRows) NextResultSet() error { - return r.Rows.(driver.RowsNextResultSet).NextResultSet() -} - -func (r *wrapperRows) ColumnTypeScanType(index int) reflect.Type { - return r.Rows.(driver.RowsColumnTypeScanType).ColumnTypeScanType(index) + return r.api.RowsNextResultSet(r.id) } func (r *wrapperRows) ColumnTypeDatabaseTypeName(index int) string { - return r.Rows.(driver.RowsColumnTypeDatabaseTypeName).ColumnTypeDatabaseTypeName(index) -} - -func (r *wrapperRows) ColumnTypeLength(index int) (length int64, ok bool) { - return r.Rows.(driver.RowsColumnTypeLength).ColumnTypeLength(index) -} - -func (r *wrapperRows) ColumnTypeNullable(index int) (nullable, ok bool) { - return r.Rows.(driver.RowsColumnTypeNullable).ColumnTypeNullable(index) + return r.api.RowsColumnTypeDatabaseTypeName(r.id, index) } func (r *wrapperRows) ColumnTypePrecisionScale(index int) (precision, scale int64, ok bool) { - return r.Rows.(driver.RowsColumnTypePrecisionScale).ColumnTypePrecisionScale(index) + return r.api.RowsColumnTypePrecisionScale(r.id, index) } diff --git a/store/sqlstore/store_test.go b/store/sqlstore/store_test.go index 3bec50df4b..55081f0663 100644 --- a/store/sqlstore/store_test.go +++ b/store/sqlstore/store_test.go @@ -4,7 +4,6 @@ package sqlstore import ( - "database/sql" "fmt" "os" "regexp" @@ -21,7 +20,6 @@ import ( "github.com/mattermost/mattermost-server/v5/einterfaces/mocks" "github.com/mattermost/mattermost-server/v5/model" - "github.com/mattermost/mattermost-server/v5/shared/driver" "github.com/mattermost/mattermost-server/v5/store" "github.com/mattermost/mattermost-server/v5/store/searchtest" "github.com/mattermost/mattermost-server/v5/store/storetest" @@ -97,47 +95,6 @@ func StoreTestWithSqlStore(t *testing.T, f func(*testing.T, store.Store, storete } } -func TestDBConnector(t *testing.T) { - testDrivers := []string{ - model.DATABASE_DRIVER_POSTGRES, - model.DATABASE_DRIVER_MYSQL, - } - - for _, dr := range testDrivers { - settings := makeSqlSettings(dr) - - store := &SqlStore{ - settings: settings, - } - connector, err := driver.NewConnector(*settings.DriverName, *settings.DataSource) - require.NoError(t, err) - db := sql.OpenDB(connector) - - store.master = getDBMap(settings, db) - store.stores.post = newSqlPostStore(store, nil) - store.stores.channel = newSqlChannelStore(store, nil) - store.stores.team = newSqlTeamStore(store) - store.stores.thread = newSqlThreadStore(store) - store.stores.user = newSqlUserStore(store, nil) - store.stores.preference = newSqlPreferenceStore(store) - store.stores.bot = newSqlBotStore(store, nil) - store.stores.scheme = newSqlSchemeStore(store) - err = store.GetMaster().CreateTablesIfNotExists() - require.NoError(t, err) - - store.stores.post.(*SqlPostStore).createIndexesIfNotExists() - store.stores.channel.(*SqlChannelStore).createIndexesIfNotExists() - - t.Run(dr, func(t *testing.T) { - // Just testing post store for now. - // This will eventually go away when it is replaced with RPC. - storetest.TestPostStore(t, store, store) - }) - - store.Close() - } -} - func initStores() { if testing.Short() { return diff --git a/utils/test_files_compiler.go b/utils/test_files_compiler.go index ff6bc14c29..77ad84ce02 100644 --- a/utils/test_files_compiler.go +++ b/utils/test_files_compiler.go @@ -43,3 +43,32 @@ func CompileGo(t *testing.T, sourceCode, outputPath string) { } require.NoError(t, err, "failed to compile go") } + +func CompileGoTest(t *testing.T, sourceCode, outputPath string) { + dir, err := ioutil.TempDir(".", "") + require.NoError(t, err) + defer os.RemoveAll(dir) + + dir, err = filepath.Abs(dir) + require.NoError(t, err) + + // Write out main.go given the source code. + main := filepath.Join(dir, "main_test.go") + err = ioutil.WriteFile(main, []byte(sourceCode), 0600) + require.NoError(t, err) + + _, sourceFile, _, ok := runtime.Caller(0) + require.True(t, ok) + serverPath := filepath.Dir(filepath.Dir(sourceFile)) + + out := &bytes.Buffer{} + cmd := exec.Command("go", "test", "-c", "-o", outputPath, main) + cmd.Dir = serverPath + cmd.Stdout = out + cmd.Stderr = out + err = cmd.Run() + if err != nil { + t.Log("Go compile errors:\n", out.String()) + } + require.NoError(t, err, "failed to compile go") +} diff --git a/web/web_test.go b/web/web_test.go index dfc13ced1c..d4862503a3 100644 --- a/web/web_test.go +++ b/web/web_test.go @@ -270,7 +270,7 @@ func TestPublicFilesRequest(t *testing.T) { defer os.RemoveAll(pluginDir) defer os.RemoveAll(webappPluginDir) - env, err := plugin.NewEnvironment(th.NewPluginAPI, pluginDir, webappPluginDir, th.App.Log(), nil) + env, err := plugin.NewEnvironment(th.NewPluginAPI, app.NewDriverImpl(th.Server), pluginDir, webappPluginDir, th.App.Log(), nil) require.NoError(t, err) pluginID := "com.mattermost.sample"