GraphQL: Unlock goroutine throttle to match batch capacity (#20146)
We were throttling the amount of concurrent resolvers at a given time. The idea behind this was to avoid overloading the database with too many requests. However, with the introduction of dataloaders, this limitation actually becomes a bottleneck because all DB calls are actually batched, so we are unnecessarily throttling the amount of items that can be processed in a single batch. The only caveat with this is that now all resolvers need to backed by dataloaders, or otherwise not be queried as part of a loop. In a subsequent PR, we will be removing channel stats from under channel to be a top-level object to be returned for a given channel. ```release-note NONE ```
Этот коммит содержится в:
коммит произвёл
GitHub
родитель
a3af488492
Коммит
fed5404166
@@ -14,6 +14,7 @@ import (
|
||||
gqlerrors "github.com/graph-gophers/graphql-go/errors"
|
||||
"github.com/mattermost/mattermost-server/v6/model"
|
||||
"github.com/mattermost/mattermost-server/v6/shared/mlog"
|
||||
"github.com/mattermost/mattermost-server/v6/web"
|
||||
)
|
||||
|
||||
type graphQLInput struct {
|
||||
@@ -22,6 +23,19 @@ type graphQLInput struct {
|
||||
Variables map[string]interface{} `json:"variables"`
|
||||
}
|
||||
|
||||
// Unique type to hold our context.
|
||||
type ctxKey int
|
||||
|
||||
const (
|
||||
webCtx ctxKey = 0
|
||||
rolesLoaderCtx ctxKey = 1
|
||||
channelsLoaderCtx ctxKey = 2
|
||||
teamsLoaderCtx ctxKey = 3
|
||||
usersLoaderCtx ctxKey = 4
|
||||
)
|
||||
|
||||
const loaderBatchCapacity = web.PerPageMaximum + 100
|
||||
|
||||
//go:embed schema.graphqls
|
||||
var schemaRaw string
|
||||
|
||||
@@ -35,7 +49,8 @@ func (api *API) InitGraphQL() error {
|
||||
opts := []graphql.SchemaOpt{
|
||||
graphql.UseFieldResolvers(),
|
||||
graphql.Logger(mlog.NewGraphQLLogger(api.srv.Log)),
|
||||
graphql.MaxParallelism(5),
|
||||
graphql.MaxParallelism(loaderBatchCapacity), // This is dangerous if the query
|
||||
// uses any non-dataloader backed object. So we need to be a bit careful here.
|
||||
}
|
||||
|
||||
if isProd() {
|
||||
@@ -58,18 +73,6 @@ func (api *API) InitGraphQL() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Unique type to hold our context.
|
||||
type ctxKey int
|
||||
|
||||
const (
|
||||
webCtx ctxKey = 0
|
||||
rolesLoaderCtx ctxKey = 1
|
||||
channelsLoaderCtx ctxKey = 2
|
||||
teamsLoaderCtx ctxKey = 3
|
||||
)
|
||||
|
||||
const loaderBatchCapacity = 200
|
||||
|
||||
func (api *API) graphQL(c *Context, w http.ResponseWriter, r *http.Request) {
|
||||
var response *graphql.Response
|
||||
defer func() {
|
||||
@@ -111,6 +114,9 @@ func (api *API) graphQL(c *Context, w http.ResponseWriter, r *http.Request) {
|
||||
teamsLoader := dataloader.NewBatchedLoader(graphQLTeamsLoader, dataloader.WithBatchCapacity(loaderBatchCapacity))
|
||||
reqCtx = context.WithValue(reqCtx, teamsLoaderCtx, teamsLoader)
|
||||
|
||||
usersLoader := dataloader.NewBatchedLoader(graphQLUsersLoader, dataloader.WithBatchCapacity(loaderBatchCapacity))
|
||||
reqCtx = context.WithValue(reqCtx, usersLoaderCtx, usersLoader)
|
||||
|
||||
response = api.schema.Exec(reqCtx,
|
||||
params.Query,
|
||||
params.OperationName,
|
||||
|
||||
@@ -313,3 +313,12 @@ func getTeamsLoader(ctx context.Context) (*dataloader.Loader, error) {
|
||||
}
|
||||
return l, nil
|
||||
}
|
||||
|
||||
// getUsersLoader returns the users loader out of the context.
|
||||
func getUsersLoader(ctx context.Context) (*dataloader.Loader, error) {
|
||||
l, ok := ctx.Value(usersLoaderCtx).(*dataloader.Loader)
|
||||
if !ok {
|
||||
return nil, errors.New("no dataloader.Loader found in context")
|
||||
}
|
||||
return l, nil
|
||||
}
|
||||
|
||||
@@ -32,16 +32,17 @@ func getGraphQLUser(ctx context.Context, id string) (*user, error) {
|
||||
return nil, web.NewInvalidParamError("user_id")
|
||||
}
|
||||
|
||||
canSee, appErr := c.App.UserCanSeeOtherUser(c.AppContext.Session().UserId, id)
|
||||
if appErr != nil || !canSee {
|
||||
c.SetPermissionError(model.PermissionViewMembers)
|
||||
return nil, c.Err
|
||||
loader, err := getUsersLoader(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
usr, appErr := c.App.GetUser(id)
|
||||
if appErr != nil {
|
||||
return nil, appErr
|
||||
thunk := loader.Load(ctx, dataloader.StringKey(id))
|
||||
result, err := thunk()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
usr := result.(*model.User)
|
||||
|
||||
if c.IsSystemAdmin() || c.AppContext.Session().UserId == usr.Id {
|
||||
userTermsOfService, appErr := c.App.GetUserTermsOfService(usr.Id)
|
||||
@@ -153,3 +154,61 @@ func (u *user) Sessions(ctx context.Context) ([]*model.Session, error) {
|
||||
|
||||
return sessions, nil
|
||||
}
|
||||
|
||||
func graphQLUsersLoader(ctx context.Context, keys dataloader.Keys) []*dataloader.Result {
|
||||
stringKeys := keys.Keys()
|
||||
result := make([]*dataloader.Result, len(stringKeys))
|
||||
|
||||
c, err := getCtx(ctx)
|
||||
if err != nil {
|
||||
for i := range result {
|
||||
result[i] = &dataloader.Result{Error: err}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
users, err := getGraphQLUsers(c, stringKeys)
|
||||
if err != nil {
|
||||
for i := range result {
|
||||
result[i] = &dataloader.Result{Error: err}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
for i, user := range users {
|
||||
result[i] = &dataloader.Result{Data: user}
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
func getGraphQLUsers(c *web.Context, userIDs []string) ([]*model.User, error) {
|
||||
// Usually this will be called only for one user
|
||||
// and cached for the rest of the query. So it's not an issue
|
||||
// to run this in a loop.
|
||||
for _, id := range userIDs {
|
||||
canSee, appErr := c.App.UserCanSeeOtherUser(c.AppContext.Session().UserId, id)
|
||||
if appErr != nil || !canSee {
|
||||
c.SetPermissionError(model.PermissionViewMembers)
|
||||
return nil, c.Err
|
||||
}
|
||||
}
|
||||
|
||||
users, appErr := c.App.GetUsers(userIDs)
|
||||
if appErr != nil {
|
||||
return nil, appErr
|
||||
}
|
||||
|
||||
// The users need to be in the exact same order as the input slice.
|
||||
tmp := make(map[string]*model.User)
|
||||
for _, u := range users {
|
||||
tmp[u.Id] = u
|
||||
}
|
||||
|
||||
// We reuse the same slice and just rewrite the roles.
|
||||
for i, uID := range userIDs {
|
||||
users[i] = tmp[uID]
|
||||
}
|
||||
|
||||
return users, nil
|
||||
}
|
||||
|
||||
@@ -769,11 +769,12 @@ type AppIface interface {
|
||||
GetUserByUsername(username string) (*model.User, *model.AppError)
|
||||
GetUserForLogin(id, loginId string) (*model.User, *model.AppError)
|
||||
GetUserTermsOfService(userID string) (*model.UserTermsOfService, *model.AppError)
|
||||
GetUsers(options *model.UserGetOptions) ([]*model.User, *model.AppError)
|
||||
GetUsers(userIDs []string) ([]*model.User, *model.AppError)
|
||||
GetUsersByGroupChannelIds(c *request.Context, channelIDs []string, asAdmin bool) (map[string][]*model.User, *model.AppError)
|
||||
GetUsersByIds(userIDs []string, options *store.UserGetByIdsOpts) ([]*model.User, *model.AppError)
|
||||
GetUsersByUsernames(usernames []string, asAdmin bool, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, *model.AppError)
|
||||
GetUsersEtag(restrictionsHash string) string
|
||||
GetUsersFromProfiles(options *model.UserGetOptions) ([]*model.User, *model.AppError)
|
||||
GetUsersInChannel(options *model.UserGetOptions) ([]*model.User, *model.AppError)
|
||||
GetUsersInChannelByStatus(options *model.UserGetOptions) ([]*model.User, *model.AppError)
|
||||
GetUsersInChannelMap(options *model.UserGetOptions, asAdmin bool) (map[string]*model.User, *model.AppError)
|
||||
|
||||
@@ -101,7 +101,7 @@ func (a *App) GetWarnMetricsBot() (*model.Bot, *model.AppError) {
|
||||
Inactive: false,
|
||||
}
|
||||
|
||||
sysAdminList, err := a.GetUsers(userOptions)
|
||||
sysAdminList, err := a.GetUsersFromProfiles(userOptions)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -130,7 +130,7 @@ func (a *App) GetSystemBot() (*model.Bot, *model.AppError) {
|
||||
Inactive: false,
|
||||
}
|
||||
|
||||
sysAdminList, err := a.GetUsers(userOptions)
|
||||
sysAdminList, err := a.GetUsersFromProfiles(userOptions)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -485,7 +485,7 @@ func (a *App) notifySysadminsBotOwnerDeactivated(c *request.Context, userID stri
|
||||
// get sysadmins
|
||||
var sysAdmins []*model.User
|
||||
for {
|
||||
sysAdminsList, err := a.GetUsers(userOptions)
|
||||
sysAdminsList, err := a.GetUsersFromProfiles(userOptions)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -19,7 +19,7 @@ func (a *App) getSysAdminsEmailRecipients() ([]*model.User, *model.AppError) {
|
||||
Role: model.SystemAdminRoleId,
|
||||
Inactive: false,
|
||||
}
|
||||
return a.GetUsers(userOptions)
|
||||
return a.GetUsersFromProfiles(userOptions)
|
||||
}
|
||||
|
||||
func (a *App) SendPaymentFailedEmail(failedPayment *model.FailedPayment) *model.AppError {
|
||||
|
||||
@@ -188,12 +188,12 @@ func TestExportAllUsers(t *testing.T) {
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 0, i)
|
||||
|
||||
users1, err := th1.App.GetUsers(&model.UserGetOptions{
|
||||
users1, err := th1.App.GetUsersFromProfiles(&model.UserGetOptions{
|
||||
Page: 0,
|
||||
PerPage: 10,
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
users2, err := th2.App.GetUsers(&model.UserGetOptions{
|
||||
users2, err := th2.App.GetUsersFromProfiles(&model.UserGetOptions{
|
||||
Page: 0,
|
||||
PerPage: 10,
|
||||
})
|
||||
@@ -202,13 +202,13 @@ func TestExportAllUsers(t *testing.T) {
|
||||
assert.ElementsMatch(t, users1, users2)
|
||||
|
||||
// Checking whether deactivated users were included in bulk export
|
||||
deletedUsers1, err := th1.App.GetUsers(&model.UserGetOptions{
|
||||
deletedUsers1, err := th1.App.GetUsersFromProfiles(&model.UserGetOptions{
|
||||
Inactive: true,
|
||||
Page: 0,
|
||||
PerPage: 10,
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
deletedUsers2, err := th1.App.GetUsers(&model.UserGetOptions{
|
||||
deletedUsers2, err := th1.App.GetUsersFromProfiles(&model.UserGetOptions{
|
||||
Inactive: true,
|
||||
Page: 0,
|
||||
PerPage: 10,
|
||||
|
||||
@@ -9970,7 +9970,7 @@ func (a *OpenTracingAppLayer) GetUserTermsOfService(userID string) (*model.UserT
|
||||
return resultVar0, resultVar1
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) GetUsers(options *model.UserGetOptions) ([]*model.User, *model.AppError) {
|
||||
func (a *OpenTracingAppLayer) GetUsers(userIDs []string) ([]*model.User, *model.AppError) {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetUsers")
|
||||
|
||||
@@ -9982,7 +9982,7 @@ func (a *OpenTracingAppLayer) GetUsers(options *model.UserGetOptions) ([]*model.
|
||||
}()
|
||||
|
||||
defer span.Finish()
|
||||
resultVar0, resultVar1 := a.app.GetUsers(options)
|
||||
resultVar0, resultVar1 := a.app.GetUsers(userIDs)
|
||||
|
||||
if resultVar1 != nil {
|
||||
span.LogFields(spanlog.Error(resultVar1))
|
||||
@@ -10075,6 +10075,28 @@ func (a *OpenTracingAppLayer) GetUsersEtag(restrictionsHash string) string {
|
||||
return resultVar0
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) GetUsersFromProfiles(options *model.UserGetOptions) ([]*model.User, *model.AppError) {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetUsersFromProfiles")
|
||||
|
||||
a.ctx = newCtx
|
||||
a.app.Srv().Store.SetContext(newCtx)
|
||||
defer func() {
|
||||
a.app.Srv().Store.SetContext(origCtx)
|
||||
a.ctx = origCtx
|
||||
}()
|
||||
|
||||
defer span.Finish()
|
||||
resultVar0, resultVar1 := a.app.GetUsersFromProfiles(options)
|
||||
|
||||
if resultVar1 != nil {
|
||||
span.LogFields(spanlog.Error(resultVar1))
|
||||
ext.Error.Set(span, true)
|
||||
}
|
||||
|
||||
return resultVar0, resultVar1
|
||||
}
|
||||
|
||||
func (a *OpenTracingAppLayer) GetUsersInChannel(options *model.UserGetOptions) ([]*model.User, *model.AppError) {
|
||||
origCtx := a.ctx
|
||||
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetUsersInChannel")
|
||||
|
||||
@@ -244,7 +244,7 @@ func (api *PluginAPI) DeleteUser(userID string) *model.AppError {
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetUsers(options *model.UserGetOptions) ([]*model.User, *model.AppError) {
|
||||
return api.app.GetUsers(options)
|
||||
return api.app.GetUsersFromProfiles(options)
|
||||
}
|
||||
|
||||
func (api *PluginAPI) GetUser(userID string) (*model.User, *model.AppError) {
|
||||
|
||||
@@ -79,7 +79,7 @@ func (cfg *AutoChannelCreator) CreateTestDMs(c *request.Context, num utils.Range
|
||||
numDMs := utils.RandIntFromRange(num)
|
||||
dms := make([]*model.Channel, numDMs)
|
||||
|
||||
users, err := cfg.a.GetUsers(&model.UserGetOptions{Page: 0, PerPage: numDMs})
|
||||
users, err := cfg.a.GetUsersFromProfiles(&model.UserGetOptions{Page: 0, PerPage: numDMs})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
13
app/user.go
13
app/user.go
@@ -380,6 +380,15 @@ func (a *App) GetUser(userID string) (*model.User, *model.AppError) {
|
||||
return user, nil
|
||||
}
|
||||
|
||||
func (a *App) GetUsers(userIDs []string) ([]*model.User, *model.AppError) {
|
||||
users, err := a.ch.srv.userService.GetUsers(userIDs)
|
||||
if err != nil {
|
||||
return nil, model.NewAppError("GetUsers", "app.user.get.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
return users, nil
|
||||
}
|
||||
|
||||
func (a *App) GetUserByUsername(username string) (*model.User, *model.AppError) {
|
||||
result, err := a.ch.srv.userService.GetUserByUsername(username)
|
||||
if err != nil {
|
||||
@@ -426,8 +435,8 @@ func (a *App) GetUserByAuth(authData *string, authService string) (*model.User,
|
||||
return user, nil
|
||||
}
|
||||
|
||||
func (a *App) GetUsers(options *model.UserGetOptions) ([]*model.User, *model.AppError) {
|
||||
users, err := a.ch.srv.userService.GetUsers(options)
|
||||
func (a *App) GetUsersFromProfiles(options *model.UserGetOptions) ([]*model.User, *model.AppError) {
|
||||
users, err := a.ch.srv.userService.GetUsersFromProfiles(options)
|
||||
if err != nil {
|
||||
return nil, model.NewAppError("GetUsers", "app.user.get_profiles.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
@@ -574,7 +574,7 @@ func TestRestrictedViewMembers(t *testing.T) {
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.Name, func(t *testing.T) {
|
||||
options := model.UserGetOptions{Page: 0, PerPage: 100, ViewRestrictions: tc.Restrictions}
|
||||
results, err := th.App.GetUsers(&options)
|
||||
results, err := th.App.GetUsersFromProfiles(&options)
|
||||
require.Nil(t, err)
|
||||
ids := []string{}
|
||||
for _, result := range results {
|
||||
|
||||
@@ -93,6 +93,10 @@ func (us *UserService) GetUser(userID string) (*model.User, error) {
|
||||
return us.store.Get(context.Background(), userID)
|
||||
}
|
||||
|
||||
func (us *UserService) GetUsers(userIDs []string) ([]*model.User, error) {
|
||||
return us.store.GetMany(context.Background(), userIDs)
|
||||
}
|
||||
|
||||
func (us *UserService) GetUserByUsername(username string) (*model.User, error) {
|
||||
return us.store.GetByUsername(username)
|
||||
}
|
||||
@@ -105,7 +109,7 @@ func (us *UserService) GetUserByAuth(authData *string, authService string) (*mod
|
||||
return us.store.GetByAuth(authData, authService)
|
||||
}
|
||||
|
||||
func (us *UserService) GetUsers(options *model.UserGetOptions) ([]*model.User, error) {
|
||||
func (us *UserService) GetUsersFromProfiles(options *model.UserGetOptions) ([]*model.User, error) {
|
||||
return us.store.GetAllProfiles(options)
|
||||
}
|
||||
|
||||
@@ -114,7 +118,7 @@ func (us *UserService) GetUsersByUsernames(usernames []string, options *model.Us
|
||||
}
|
||||
|
||||
func (us *UserService) GetUsersPage(options *model.UserGetOptions, asAdmin bool) ([]*model.User, error) {
|
||||
users, err := us.GetUsers(options)
|
||||
users, err := us.GetUsersFromProfiles(options)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
Ссылка в новой задаче
Block a user