mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
fix: improve http connection pooling for smtp notifications (#20605)
This change updates how SMTP notifications are polled during scale tests. Before, each of the ~2,000 pollers created its own http.Client, which opened thousands of short-lived TCP connections. Under heavy load, this ran out of available network ports and caused errors like `connect: cannot assign requested address` Now, all pollers share one HTTP connection pool. This prevents port exhaustion and makes polling faster and more stable. If a network error happens, the poller will now retry instead of stopping, so tests keep running until all notifications are received. The `SMTPRequestTimeout` is now applied per request using a context, instead of being set on the `http.Client`.
This commit is contained in:
@@ -142,6 +142,15 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command {
|
||||
triggerTimes[id] = make(chan time.Time, 1)
|
||||
}
|
||||
|
||||
smtpHTTPTransport := &http.Transport{
|
||||
MaxConnsPerHost: 512,
|
||||
MaxIdleConnsPerHost: 512,
|
||||
IdleConnTimeout: 60 * time.Second,
|
||||
}
|
||||
smtpHTTPClient := &http.Client{
|
||||
Transport: smtpHTTPTransport,
|
||||
}
|
||||
|
||||
configs := make([]notifications.Config, 0, userCount)
|
||||
for range templateAdminCount {
|
||||
config := notifications.Config{
|
||||
@@ -157,6 +166,7 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command {
|
||||
Metrics: metrics,
|
||||
SMTPApiURL: smtpAPIURL,
|
||||
SMTPRequestTimeout: smtpRequestTimeout,
|
||||
SMTPHttpClient: smtpHTTPClient,
|
||||
}
|
||||
if err := config.Validate(); err != nil {
|
||||
return xerrors.Errorf("validate config: %w", err)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package notifications
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -40,6 +41,9 @@ type Config struct {
|
||||
|
||||
// SMTPRequestTimeout is the timeout for SMTP requests.
|
||||
SMTPRequestTimeout time.Duration `json:"smtp_request_timeout"`
|
||||
|
||||
// SMTPHttpClient is the HTTP client for SMTP requests.
|
||||
SMTPHttpClient *http.Client `json:"-"`
|
||||
}
|
||||
|
||||
func (c Config) Validate() error {
|
||||
@@ -68,6 +72,10 @@ func (c Config) Validate() error {
|
||||
return xerrors.New("smtp_request_timeout must be set if smtp_api_url is set")
|
||||
}
|
||||
|
||||
if c.SMTPApiURL != "" && c.SMTPHttpClient == nil {
|
||||
return xerrors.New("smtp_http_client must be set if smtp_api_url is set")
|
||||
}
|
||||
|
||||
if c.DialTimeout <= 0 {
|
||||
return xerrors.New("dial_timeout must be greater than 0")
|
||||
}
|
||||
|
||||
@@ -298,15 +298,16 @@ func (r *Runner) watchNotificationsSMTP(ctx context.Context, user codersdk.User,
|
||||
receivedNotifications := make(map[uuid.UUID]struct{})
|
||||
|
||||
apiURL := fmt.Sprintf("%s/messages?email=%s", r.cfg.SMTPApiURL, user.Email)
|
||||
httpClient := &http.Client{
|
||||
Timeout: r.cfg.SMTPRequestTimeout,
|
||||
}
|
||||
httpClient := r.cfg.SMTPHttpClient
|
||||
|
||||
const smtpPollInterval = 2 * time.Second
|
||||
done := xerrors.New("done")
|
||||
|
||||
tkr := r.clock.TickerFunc(ctx, smtpPollInterval, func() error {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil)
|
||||
reqCtx, cancel := context.WithTimeout(ctx, r.cfg.SMTPRequestTimeout)
|
||||
defer cancel()
|
||||
|
||||
req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, apiURL, nil)
|
||||
if err != nil {
|
||||
logger.Error(ctx, "create SMTP API request", slog.Error(err))
|
||||
r.cfg.Metrics.AddError("smtp_create_request")
|
||||
@@ -317,14 +318,16 @@ func (r *Runner) watchNotificationsSMTP(ctx context.Context, user codersdk.User,
|
||||
if err != nil {
|
||||
logger.Error(ctx, "poll smtp api for notifications", slog.Error(err))
|
||||
r.cfg.Metrics.AddError("smtp_poll")
|
||||
return xerrors.Errorf("poll smtp api: %w", err)
|
||||
return nil
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
// discard the response to allow reusing of the connection
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
_ = resp.Body.Close()
|
||||
logger.Error(ctx, "smtp api returned non-200 status", slog.F("status", resp.StatusCode))
|
||||
r.cfg.Metrics.AddError("smtp_bad_status")
|
||||
return xerrors.Errorf("smtp api returned status %d", resp.StatusCode)
|
||||
return nil
|
||||
}
|
||||
|
||||
var summaries []smtpmock.EmailSummary
|
||||
|
||||
@@ -212,6 +212,8 @@ func TestRunWithSMTP(t *testing.T) {
|
||||
smtpTrap := mClock.Trap().TickerFunc("smtp")
|
||||
defer smtpTrap.Close()
|
||||
|
||||
httpClient := &http.Client{}
|
||||
|
||||
// Start receiving runners who will receive notifications
|
||||
receivingRunners := make([]*notifications.Runner, 0, numReceivingUsers)
|
||||
for i := range numReceivingUsers {
|
||||
@@ -229,6 +231,7 @@ func TestRunWithSMTP(t *testing.T) {
|
||||
ExpectedNotificationsIDs: expectedNotificationsIDs,
|
||||
SMTPApiURL: smtpAPIServer.URL,
|
||||
SMTPRequestTimeout: testutil.WaitLong,
|
||||
SMTPHttpClient: httpClient,
|
||||
}
|
||||
err := runnerCfg.Validate()
|
||||
require.NoError(t, err)
|
||||
|
||||
Reference in New Issue
Block a user