mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

Pluginstore migration (#14971)

Просмотр исходного кода 
Automatic Merge
Этот коммит содержится в:
Rodrigo Villablanca
2020-08-13 01:35:57 -04:00
коммит произвёл GitHub
родитель 19173ea6ff
Коммит 909cda6d49
11 изменённых файлов: 363 добавлений и 236 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

93
store/sqlstore/plugin_store.go
Просмотреть файл

@@ -7,9 +7,9 @@ import (
"bytes"
"database/sql"
"fmt"
"net/http"
sq "github.com/Masterminds/squirrel"
"github.com/pkg/errors"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/store"
@@ -39,7 +39,7 @@ func newSqlPluginStore(sqlStore SqlStore) store.PluginStore {
func (ps SqlPluginStore) createIndexesIfNotExists() {
}
func (ps SqlPluginStore) SaveOrUpdate(kv *model.PluginKeyValue) (*model.PluginKeyValue, *model.AppError) {
func (ps SqlPluginStore) SaveOrUpdate(kv *model.PluginKeyValue) (*model.PluginKeyValue, error) {
if err := kv.IsValid(); err != nil {
return nil, err
}
@@ -58,11 +58,11 @@ func (ps SqlPluginStore) SaveOrUpdate(kv *model.PluginKeyValue) (*model.PluginKe
// Unfortunately PostgreSQL pre-9.5 does not have an atomic upsert, so we use
// separate update and insert queries to accomplish our upsert
if rowsAffected, err := ps.GetMaster().Update(kv); err != nil {
return nil, model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return nil, errors.Wrap(err, "failed to update PluginKeyValue")
} else if rowsAffected == 0 {
// No rows were affected by the update, so let's try an insert
if err := ps.GetMaster().Insert(kv); err != nil {
return nil, model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusBadRequest)
return nil, errors.Wrap(err, "failed to save PluginKeyValue")
}
}
} else if ps.DriverName() == model.DATABASE_DRIVER_MYSQL {
@@ -74,18 +74,18 @@ func (ps SqlPluginStore) SaveOrUpdate(kv *model.PluginKeyValue) (*model.PluginKe
queryString, args, err := query.ToSql()
if err != nil {
return nil, model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql.build_query.app_error", nil, err.Error(), http.StatusInternalServerError)
return nil, errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return nil, model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return nil, errors.Wrap(err, "failed to upsert PluginKeyValue")
}
}
return kv, nil
}
func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte) (bool, *model.AppError) {
func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte) (bool, error) {
if err := kv.IsValid(); err != nil {
return false, err
}
@@ -106,11 +106,11 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
queryString, args, err := query.ToSql()
if err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql.build_query.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql_plugin_store.delete.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "failed to delete PluginKeyValue")
}
// Insert if oldValue is nil
@@ -121,7 +121,7 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
if IsUniqueConstraintError(err, []string{"PRIMARY", "PluginId", "Key", "PKey", "pkey"}) {
return false, nil
} else {
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "failed to insert PluginKeyValue")
}
}
} else {
@@ -142,17 +142,17 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
queryString, args, err := query.ToSql()
if err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql.build_query.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "plugin_tosql")
}
updateResult, err := ps.GetMaster().Exec(queryString, args...)
if err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "failed to update PluginKeyValue")
}
if rowsAffected, err := updateResult.RowsAffected(); err != nil {
// Failed to update
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "unable to get rows affected")
} else if rowsAffected == 0 {
if ps.DriverName() == model.DATABASE_DRIVER_MYSQL && bytes.Equal(oldValue, kv.Value) {
// ROW_COUNT on MySQL is zero even if the row existed but no changes to the row were required.
@@ -173,12 +173,12 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
queryString, args, err := query.ToSql()
if err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql.build_query.app_error", nil, fmt.Sprintf("plugin_id=%v, key=%v, err=%v", kv.PluginId, kv.Key, err.Error()), http.StatusInternalServerError)
return false, errors.Wrap(err, "plugin_tosql")
}
count, err := ps.GetReplica().SelectInt(queryString, args...)
if err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql_plugin_store.compare_and_set.mysql_select.app_error", nil, fmt.Sprintf("plugin_id=%v, key=%v, err=%v", kv.PluginId, kv.Key, err.Error()), http.StatusInternalServerError)
return false, errors.Wrapf(err, "failed to count PluginKeyValue with pluginId=%s and key=%s", kv.PluginId, kv.Key)
}
if count == 0 {
@@ -186,7 +186,7 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
} else if count == 1 {
return true, nil
} else {
return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql_plugin_store.compare_and_set.too_many_rows.app_error", nil, fmt.Sprintf("plugin_id=%v, key=%v, count=%d", kv.PluginId, kv.Key, count), http.StatusInternalServerError)
return false, errors.Wrapf(err, "got too many rows when counting PluginKeyValue with pluginId=%s, key=%s, rows=%d", kv.PluginId, kv.Key, count)
}
}
@@ -199,7 +199,7 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
return true, nil
}
func (ps SqlPluginStore) CompareAndDelete(kv *model.PluginKeyValue, oldValue []byte) (bool, *model.AppError) {
func (ps SqlPluginStore) CompareAndDelete(kv *model.PluginKeyValue, oldValue []byte) (bool, error) {
if err := kv.IsValid(); err != nil {
return false, err
}
@@ -221,16 +221,16 @@ func (ps SqlPluginStore) CompareAndDelete(kv *model.PluginKeyValue, oldValue []b
queryString, args, err := query.ToSql()
if err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndDelete", "store.sql.build_query.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "plugin_tosql")
}
deleteResult, err := ps.GetMaster().Exec(queryString, args...)
if err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndDelete", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "failed to delete PluginKeyValue")
}
if rowsAffected, err := deleteResult.RowsAffected(); err != nil {
return false, model.NewAppError("SqlPluginStore.CompareAndDelete", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return false, errors.Wrap(err, "unable to get rows affected")
} else if rowsAffected == 0 {
return false, nil
}
@@ -238,7 +238,7 @@ func (ps SqlPluginStore) CompareAndDelete(kv *model.PluginKeyValue, oldValue []b
return true, nil
}
func (ps SqlPluginStore) SetWithOptions(pluginId string, key string, value []byte, opt model.PluginKVSetOptions) (bool, *model.AppError) {
func (ps SqlPluginStore) SetWithOptions(pluginId string, key string, value []byte, opt model.PluginKVSetOptions) (bool, error) {
if err := opt.IsValid(); err != nil {
return false, err
}
@@ -252,27 +252,16 @@ func (ps SqlPluginStore) SetWithOptions(pluginId string, key string, value []byt
return ps.CompareAndSet(kv, opt.OldValue)
}
savedKv, err := ps.SaveOrUpdate(kv)
if err != nil {
return false, err
savedKv, nErr := ps.SaveOrUpdate(kv)
if nErr != nil {
return false, nErr
}
return savedKv != nil, nil
}
func (ps SqlPluginStore) Get(pluginId, key string) (*model.PluginKeyValue, *model.AppError) {
func (ps SqlPluginStore) Get(pluginId, key string) (*model.PluginKeyValue, error) {
currentTime := model.GetMillis()
failure := func(err error, statusCode int) *model.AppError {
return model.NewAppError(
"SqlPluginStore.Get",
"store.sql_plugin_store.get.app_error",
nil,
fmt.Sprintf("plugin_id=%v, key=%v, err=%v", pluginId, key, err.Error()),
statusCode,
)
}
query := ps.getQueryBuilder().Select("PluginId, PKey, PValue, ExpireAt").
From("PluginKeyValueStore").
Where(sq.Eq{"PluginId": pluginId}).
@@ -280,21 +269,22 @@ func (ps SqlPluginStore) Get(pluginId, key string) (*model.PluginKeyValue, *mode
Where(sq.Or{sq.Eq{"ExpireAt": 0}, sq.Gt{"ExpireAt": currentTime}})
queryString, args, err := query.ToSql()
if err != nil {
return nil, failure(err, http.StatusInternalServerError)
return nil, errors.Wrap(err, "plugin_tosql")
}
row := ps.GetReplica().Db.QueryRow(queryString, args...)
var kv model.PluginKeyValue
if err := row.Scan(&kv.PluginId, &kv.Key, &kv.Value, &kv.ExpireAt); err != nil {
if err == sql.ErrNoRows {
return nil, failure(err, http.StatusNotFound)
return nil, store.NewErrNotFound("PluginKeyValue", fmt.Sprintf("pluginId=%s, key=%s", pluginId, key))
}
return nil, failure(err, http.StatusInternalServerError)
return nil, errors.Wrapf(err, "failed to get PluginKeyValue with pluginId=%s and key=%s", pluginId, key)
}
return &kv, nil
}
func (ps SqlPluginStore) Delete(pluginId, key string) *model.AppError {
func (ps SqlPluginStore) Delete(pluginId, key string) error {
query := ps.getQueryBuilder().
Delete("PluginKeyValueStore").
Where(sq.Eq{"PluginId": pluginId}).
@@ -302,34 +292,33 @@ func (ps SqlPluginStore) Delete(pluginId, key string) *model.AppError {
queryString, args, err := query.ToSql()
if err != nil {
return model.NewAppError("SqlPluginStore.Delete", "store.sql.build_query.app_error", nil, fmt.Sprintf("plugin_id=%v, key=%v, err=%v", pluginId, key, err.Error()), http.StatusInternalServerError)
return errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return model.NewAppError("SqlPluginStore.Delete", "store.sql_plugin_store.delete.app_error", nil, fmt.Sprintf("plugin_id=%v, key=%v, err=%v", pluginId, key, err.Error()), http.StatusInternalServerError)
return errors.Wrapf(err, "failed to delete PluginKeyValue with pluginId=%s and key=%s", pluginId, key)
}
return nil
}
func (ps SqlPluginStore) DeleteAllForPlugin(pluginId string) *model.AppError {
func (ps SqlPluginStore) DeleteAllForPlugin(pluginId string) error {
query := ps.getQueryBuilder().
Delete("PluginKeyValueStore").
Where(sq.Eq{"PluginId": pluginId})
queryString, args, err := query.ToSql()
if err != nil {
return model.NewAppError("SqlPluginStore.Delete", "store.sql.build_query.app_error", nil, fmt.Sprintf("plugin_id=%v, err=%v", pluginId, err.Error()), http.StatusInternalServerError)
return errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return model.NewAppError("SqlPluginStore.Delete", "store.sql_plugin_store.delete.app_error", nil, fmt.Sprintf("plugin_id=%v, err=%v", pluginId, err.Error()), http.StatusInternalServerError)
return errors.Wrapf(err, "failed to get all PluginKeyValues with pluginId=%s ", pluginId)
}
return nil
}
func (ps SqlPluginStore) DeleteAllExpired() *model.AppError {
func (ps SqlPluginStore) DeleteAllExpired() error {
currentTime := model.GetMillis()
query := ps.getQueryBuilder().
Delete("PluginKeyValueStore").
Where(sq.NotEq{"ExpireAt": 0}).
@@ -337,16 +326,16 @@ func (ps SqlPluginStore) DeleteAllExpired() *model.AppError {
queryString, args, err := query.ToSql()
if err != nil {
return model.NewAppError("SqlPluginStore.Delete", "store.sql.build_query.app_error", nil, fmt.Sprintf("current_time=%v, err=%v", currentTime, err.Error()), http.StatusInternalServerError)
return errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return model.NewAppError("SqlPluginStore.Delete", "store.sql_plugin_store.delete.app_error", nil, fmt.Sprintf("current_time=%v, err=%v", currentTime, err.Error()), http.StatusInternalServerError)
return errors.Wrap(err, "failed to delete all expired PluginKeyValues")
}
return nil
}
func (ps SqlPluginStore) List(pluginId string, offset int, limit int) ([]string, *model.AppError) {
func (ps SqlPluginStore) List(pluginId string, offset int, limit int) ([]string, error) {
if limit <= 0 {
limit = defaultPluginKeyFetchLimit
}
@@ -371,12 +360,12 @@ func (ps SqlPluginStore) List(pluginId string, offset int, limit int) ([]string,
queryString, args, err := query.ToSql()
if err != nil {
return nil, model.NewAppError("SqlPluginStore.List", "store.sql.build_query.app_error", nil, fmt.Sprintf("plugin_id=%v, err=%v", pluginId, err.Error()), http.StatusInternalServerError)
return nil, errors.Wrap(err, "plugin_tosql")
}
_, err = ps.GetReplica().Select(&keys, queryString, args...)
if err != nil {
return nil, model.NewAppError("SqlPluginStore.List", "store.sql_plugin_store.list.app_error", nil, fmt.Sprintf("plugin_id=%v, err=%v", pluginId, err.Error()), http.StatusInternalServerError)
return nil, errors.Wrapf(err, "failed to get PluginKeyValues with pluginId=%s", pluginId)
}
return keys, nil