From 45ba1dc196c20b2ec8935cd55a81ee7759b3f8f4 Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Wed, 13 Dec 2023 09:39:22 +0530 Subject: [PATCH] MM-55655: Fix unbounded concurrency in outgoing webhooks (#25511) We were simply spawning goroutines within goroutines. For each post, we would spawn one goroutine per hook, and then one goroutine per callback URL within that hook. And to top it off, this whole thing was itself within a goroutine. To fix it, we remove the goroutine spawning at a per hook level. And then use a waitgroup to wait until all hooks from each callback URL is complete. While here, some other optimizations that we do: 1. We already had the channel object, but inspite of that, we were calling channel.get again in CreatePostMissingChannel. We just use CreatePost now and pass the channel. 2. We pre-compile the regex. 3. We store the http.Client in the server to reuse TCP connections. https://mattermost.atlassian.net/browse/MM-55655 ```release-note NONE ``` --- server/channels/app/command.go | 2 +- server/channels/app/server.go | 2 ++ server/channels/app/webhook.go | 22 ++++++++++++---------- 3 files changed, 15 insertions(+), 11 deletions(-) diff --git a/server/channels/app/command.go b/server/channels/app/command.go index ad756f4ef5..da327c8aa4 100644 --- a/server/channels/app/command.go +++ b/server/channels/app/command.go @@ -508,7 +508,7 @@ func (a *App) DoCommandRequest(cmd *model.Command, p url.Values) (*model.Command } // Send the request - resp, err := a.HTTPService().MakeClient(false).Do(req) + resp, err := a.Srv().outgoingWebhookClient.Do(req) if err != nil { return cmd, nil, model.NewAppError("command", "api.command.execute_command.failed.app_error", map[string]any{"Trigger": cmd.Trigger}, "", http.StatusInternalServerError).Wrap(err) } diff --git a/server/channels/app/server.go b/server/channels/app/server.go index 4412c328b9..4a6f59ec22 100644 --- a/server/channels/app/server.go +++ b/server/channels/app/server.go @@ -110,6 +110,7 @@ type Server struct { httpService httpservice.HTTPService PushNotificationsHub PushNotificationsHub pushNotificationClient *http.Client // TODO: move this to it's own package + outgoingWebhookClient *http.Client runEssentialJobs bool Jobs *jobs.JobServer @@ -337,6 +338,7 @@ func NewServer(options ...Option) (*Server, error) { } s.pushNotificationClient = s.httpService.MakeClient(true) + s.outgoingWebhookClient = s.httpService.MakeClient(false) if err2 := utils.TranslationsPreInit(); err2 != nil { return nil, errors.Wrapf(err2, "unable to load Mattermost translation files") diff --git a/server/channels/app/webhook.go b/server/channels/app/webhook.go index b804b6b97e..4d5385f063 100644 --- a/server/channels/app/webhook.go +++ b/server/channels/app/webhook.go @@ -12,6 +12,7 @@ import ( "net/http" "regexp" "strings" + "sync" "unicode/utf8" "github.com/mattermost/mattermost/server/public/model" @@ -28,6 +29,8 @@ const ( MaxIntegrationResponseSize = 1024 * 1024 // Posts can be <100KB at most, so this is likely more than enough ) +var linkWithTextRegex = regexp.MustCompile(`<([^\n<\|>]+)\|([^\|\n>]+)>`) + func (a *App) handleWebhookEvents(c request.CTX, post *model.Post, team *model.Team, channel *model.Channel, user *model.User) *model.AppError { if !*a.Config().ServiceSettings.EnableOutgoingWebhooks { return nil @@ -84,11 +87,7 @@ func (a *App) handleWebhookEvents(c request.CTX, post *model.Post, team *model.T TriggerWord: triggerWord, FileIds: strings.Join(post.FileIds, ","), } - a.Srv().Go(func(hook *model.OutgoingWebhook) func() { - return func() { - a.TriggerWebhook(c, payload, hook, post, channel) - } - }(hook)) + a.TriggerWebhook(c, payload, hook, post, channel) } return nil @@ -109,11 +108,14 @@ func (a *App) TriggerWebhook(c request.CTX, payload *model.OutgoingWebhookPayloa contentType = "application/x-www-form-urlencoded" } + var wg sync.WaitGroup for i := range hook.CallbackURLs { + wg.Add(1) // Get the callback URL by index to properly capture it for the go func url := hook.CallbackURLs[i] - a.Srv().Go(func() { + go func() { + defer wg.Done() webhookResp, err := a.doOutgoingWebhookRequest(url, body, contentType) if err != nil { c.Logger().Error("Event POST failed.", mlog.Err(err)) @@ -150,8 +152,9 @@ func (a *App) TriggerWebhook(c request.CTX, payload *model.OutgoingWebhookPayloa c.Logger().Error("Failed to create response post.", mlog.Err(err)) } } - }) + }() } + wg.Wait() } func (a *App) doOutgoingWebhookRequest(url string, body io.Reader, contentType string) (*model.OutgoingWebhookResponse, error) { @@ -163,7 +166,7 @@ func (a *App) doOutgoingWebhookRequest(url string, body io.Reader, contentType s req.Header.Set("Content-Type", contentType) req.Header.Set("Accept", "application/json") - resp, err := a.HTTPService().MakeClient(false).Do(req) + resp, err := a.Srv().outgoingWebhookClient.Do(req) if err != nil { return nil, err } @@ -264,7 +267,6 @@ func SplitWebhookPost(post *model.Post, maxPostSize int) ([]*model.Post, *model. func (a *App) CreateWebhookPost(c request.CTX, userID string, channel *model.Channel, text, overrideUsername, overrideIconURL, overrideIconEmoji string, props model.StringInterface, postType string, postRootId string) (*model.Post, *model.AppError) { // parse links into Markdown format - linkWithTextRegex := regexp.MustCompile(`<([^\n<\|>]+)\|([^\|\n>]+)>`) text = linkWithTextRegex.ReplaceAllString(text, "[${2}](${1})") post := &model.Post{UserId: userID, ChannelId: channel.Id, Message: text, Type: postType, RootId: postRootId} @@ -314,7 +316,7 @@ func (a *App) CreateWebhookPost(c request.CTX, userID string, channel *model.Cha } for _, split := range splits { - if _, err = a.CreatePostMissingChannel(c, split, false, false); err != nil { + if _, err = a.CreatePost(c, split, channel, false, false); err != nil { return nil, model.NewAppError("CreateWebhookPost", "api.post.create_webhook_post.creating.app_error", nil, "err="+err.Message, http.StatusInternalServerError) } }