diff --git a/cli/exp_scaletest_notifications.go b/cli/exp_scaletest_notifications.go index 6b765bc7d61..06f21003424 100644 --- a/cli/exp_scaletest_notifications.go +++ b/cli/exp_scaletest_notifications.go @@ -5,9 +5,13 @@ package cli import ( "bytes" "context" + "errors" "fmt" + "io" "net/http" + "os" "os/signal" + "slices" "strconv" "strings" "sync" @@ -16,11 +20,13 @@ import ( "github.com/google/uuid" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" + "golang.org/x/sync/errgroup" "golang.org/x/xerrors" "cdr.dev/slog/v3" notificationsLib "github.com/coder/coder/v2/coderd/notifications" "github.com/coder/coder/v2/codersdk" + "github.com/coder/coder/v2/cryptorand" "github.com/coder/coder/v2/scaletest/createusers" "github.com/coder/coder/v2/scaletest/harness" "github.com/coder/coder/v2/scaletest/loadtestutil" @@ -28,29 +34,54 @@ import ( "github.com/coder/serpent" ) +// notificationsPrefix prefixes every artifact this command names: the trigger +// template and the minted API tokens. Residue from a hard kill is greppable by it, +// and it is a subset of what "coder exp scaletest cleanup" reclaims. +const notificationsPrefix = loadtestutil.ScaleTestPrefix + "-notifications-" + +// runArtifactName returns the name this run gives its artifacts. One name covers +// the template and the tokens: both belong to the run, and one string means one +// thing to grep for. +func runArtifactName(runID string) string { + return notificationsPrefix + runID +} + +// scaletestIdleConnTimeout keeps pooled connections warm across the setup and +// SMTP request phases instead of re-dialing per request. +const scaletestIdleConnTimeout = 60 * time.Second + +// defaultPhaseTimeout bounds the setup and cleanup phases. Both flags take their +// default from here so they cannot drift apart: the two phases do the same amount +// of per-user work, so a value that suits one suits the other. +const defaultPhaseTimeout = 30 * time.Minute + func (r *RootCmd) scaletestNotifications() *serpent.Command { var ( userCount int64 templateAdminPercentage float64 - notificationTimeout time.Duration + setupConcurrency int64 + setupTimeout time.Duration + testTimeout time.Duration smtpRequestTimeout time.Duration dialTimeout time.Duration + cleanupTimeout time.Duration noCleanup bool smtpAPIURL string - tracingFlags = &scaletestTracingFlags{} - - // This test requires unlimited concurrency. - timeoutStrategy = &timeoutFlags{} - cleanupStrategy = newScaletestCleanupStrategy() + tracingFlags = &scaletestTracingFlags{} output = &scaletestOutputFlags{} prometheusFlags = &scaletestPrometheusFlags{} ) cmd := &serpent.Command{ Use: "notifications", - Short: "Simulate notification delivery by creating many users listening to notifications.", - Handler: func(inv *serpent.Invocation) error { + Short: "Simulate notification delivery by connecting many users that listen for notifications.", + Long: "This creates dedicated scaletest users and deletes them afterwards.\n" + + "\n" + + "Run only one instance of this command against a deployment at a time. Running it\n" + + "alongside another scaletest command that creates or deletes scaletest users can\n" + + "remove the users this run is connected as.", + Handler: func(inv *serpent.Invocation) (retErr error) { ctx := inv.Context() client, err := r.InitClient(inv) if err != nil { @@ -70,8 +101,42 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { return xerrors.Errorf("--user-count must be greater than 0") } - if templateAdminPercentage < 0 || templateAdminPercentage > 100 { - return xerrors.Errorf("--template-admin-percentage must be between 0 and 100") + // 0 is rejected rather than clamped: with no template admins nothing watches + // for the notification, so the run would record zero latencies and exit 0. + if templateAdminPercentage <= 0 || templateAdminPercentage > 100 { + return xerrors.Errorf("--template-admin-percentage must be greater than 0 and at most 100") + } + + // Unlike the shared scaletest concurrency flags, 0 is not "unlimited" + // here: setup and cleanup are deliberately bounded so they cannot open a + // connection per user. + if setupConcurrency <= 0 { + return xerrors.Errorf("--setup-concurrency must be greater than 0") + } + + if setupTimeout <= 0 { + return xerrors.Errorf("--setup-timeout must be greater than 0") + } + + if cleanupTimeout <= 0 { + return xerrors.Errorf("--cleanup-timeout must be greater than 0") + } + + if testTimeout <= 0 { + return xerrors.Errorf("--timeout must be greater than 0") + } + + if dialTimeout <= 0 { + return xerrors.Errorf("--dial-timeout must be greater than 0") + } + + // The connect phase must finish inside the overall budget, or the last runner + // releases the dial barrier with no time left to trigger and observe. This + // rejects only the degenerate case where connecting may consume the entire + // budget; values close to it still leave little room, so pick a dial timeout + // well under --timeout. + if dialTimeout >= testTimeout { + return xerrors.Errorf("--dial-timeout (%s) must be less than --timeout (%s), leaving time to trigger and deliver notifications", dialTimeout, testTimeout) } if smtpAPIURL != "" && !strings.HasPrefix(smtpAPIURL, "http://") && !strings.HasPrefix(smtpAPIURL, "https://") { @@ -86,8 +151,11 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { _, _ = fmt.Fprintf(inv.Stderr, "Distribution plan:\n") _, _ = fmt.Fprintf(inv.Stderr, " Total users: %d\n", userCount) - _, _ = fmt.Fprintf(inv.Stderr, " Template admins: %d (%.1f%%)\n", templateAdminCount, templateAdminPercentage) - _, _ = fmt.Fprintf(inv.Stderr, " Regular users: %d (%.1f%%)\n", regularUserCount, 100.0-templateAdminPercentage) + // Report the split actually used, which the minimum-one bump above may have + // moved away from the requested percentage. + actualAdminPercentage := float64(templateAdminCount) / float64(userCount) * 100 + _, _ = fmt.Fprintf(inv.Stderr, " Template admins: %d (%.1f%%)\n", templateAdminCount, actualAdminPercentage) + _, _ = fmt.Fprintf(inv.Stderr, " Regular users: %d (%.1f%%)\n", regularUserCount, 100.0-actualAdminPercentage) outputs, err := output.parse() if err != nil { @@ -117,13 +185,6 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { <-time.After(prometheusFlags.Wait) }() - _, _ = fmt.Fprintln(inv.Stderr, "Creating users...") - - dialBarrier := &sync.WaitGroup{} - templateAdminWatchBarrier := &sync.WaitGroup{} - dialBarrier.Add(int(userCount)) - templateAdminWatchBarrier.Add(int(templateAdminCount)) - expectedNotificationIDs := map[uuid.UUID]struct{}{ notificationsLib.TemplateTemplateDeleted: {}, } @@ -136,74 +197,184 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { smtpHTTPTransport := &http.Transport{ MaxConnsPerHost: 512, MaxIdleConnsPerHost: 512, - IdleConnTimeout: 60 * time.Second, + IdleConnTimeout: scaletestIdleConnTimeout, } smtpHTTPClient := &http.Client{ Transport: smtpHTTPTransport, } - configs := make([]notifications.Config, 0, userCount) - for range templateAdminCount { - config := notifications.Config{ - User: createusers.Config{ - OrganizationID: me.OrganizationIDs[0], - }, - Roles: []string{codersdk.RoleTemplateAdmin}, - NotificationTimeout: notificationTimeout, - DialTimeout: dialTimeout, - DialBarrier: dialBarrier, - ReceivingWatchBarrier: templateAdminWatchBarrier, - ExpectedNotificationsIDs: expectedNotificationIDs, - Metrics: metrics, - SMTPApiURL: smtpAPIURL, - SMTPRequestTimeout: smtpRequestTimeout, - SMTPHttpClient: smtpHTTPClient, - } - if err := config.Validate(); err != nil { - return xerrors.Errorf("validate config: %w", err) - } - configs = append(configs, config) + // One HTTP client for every websocket handshake. It carries the CLI's TLS and + // proxy configuration, which the websocket library would otherwise ignore in + // favor of http.DefaultClient. Deliberately not the setup client: that one + // caps MaxConnsPerHost to the setup concurrency, which would throttle the + // dials this test exists to make. A handshake hands its TCP connection to the + // caller and never returns it to the pool, so sharing one client still gives + // every runner its own connection. + dialClient, err := loadtestutil.DupClientCopyingHeaders(client, BypassHeader) + if err != nil { + return xerrors.Errorf("create dial client: %w", err) } - for range regularUserCount { + dialHTTPClient := dialClient.HTTPClient + + // Setup and cleanup run at most setupConcurrency requests at a time, so size + // the pool to match and keep connections warm instead of re-dialing per + // request. + // + // This bounds each client separately. The create path also has createusers + // duplicate the client per login, so it briefly holds more sockets than this + // number suggests. + concurrency := int(setupConcurrency) + setupClient, err := loadtestutil.DupClientConfiguringTransport(client, BypassHeader, boundPool(concurrency)) + if err != nil { + return xerrors.Errorf("create setup client: %w", err) + } + orgID := me.OrganizationIDs[0] + + // Identify everything this run creates so residue from a hard kill is + // greppable, and bound the token lifetime to a little beyond the run so + // orphans expire in hours rather than the deployment default. + runID, err := cryptorand.String(8) + if err != nil { + return xerrors.Errorf("generate run id: %w", err) + } + runName := runArtifactName(runID) + + run := &scaletestRun{ + logger: logger, + stderr: inv.Stderr, + metrics: metrics, + client: client, + setupClient: setupClient, + orgID: orgID, + concurrency: concurrency, + noCleanup: noCleanup, + userCount: int(userCount), + adminCount: int(templateAdminCount), + templateName: runName, + } + + // Always clean up whatever the run created, even on interrupt, timeout, or a + // later failure. Registered before setup starts so a partial setup is torn + // down too, and detached from ctx so the interrupt that ended the run does + // not also kill the cleanup. + if !noCleanup { + defer func() { + // Tell the operator what is happening and how to leave: killing the + // process is the one exit that leaves promoted users, live tokens, and a + // stranded template behind. + _, _ = fmt.Fprintf(inv.Stderr, + "\nCleaning up %d users and the trigger template, bounded by --cleanup-timeout=%s.\n"+ + "This can take several minutes at scale.\n"+ + "Please do not kill this process: users may be left with the template-admin\n"+ + "role and live API tokens. Interrupt again to abort cleanup deliberately.\n", + len(run.users), cleanupTimeout) + + cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), cleanupTimeout) + defer cleanupCancel() + + // A second interrupt aborts the cleanup. Without this an operator who + // believes a slow cleanup is stuck has only SIGKILL, which is worse: it + // stops cleanup at an arbitrary point with no way to report what remains. + abort := make(chan os.Signal, 1) + signal.Notify(abort, StopSignals...) + defer signal.Stop(abort) + go func() { + select { + case <-abort: + _, _ = fmt.Fprintf(inv.Stderr, + "\nAborting cleanup. Residue from this run is named %q:\n"+ + " API tokens: token name %q on the affected users\n"+ + " Template: %q in organization %s\n"+ + " Users: may still hold the %s role\n"+ + "Run \"coder exp scaletest cleanup\" to remove scaletest users, or revoke the tokens by name.\n", + runName, runName, runName, orgID, codersdk.RoleTemplateAdmin) + cleanupCancel() + case <-cleanupCtx.Done(): + } + }() + + if cerr := run.cleanup(cleanupCtx); cerr != nil { + logger.Error(ctx, "failed to clean up", slog.Error(cerr)) + retErr = errors.Join(retErr, xerrors.Errorf("clean up: %w", cerr)) + } + }() + } + + // Setup is all-or-nothing: on any failure, the deferred cleanup tears down + // whatever it managed to create before the run aborts. + setupCtx, setupCancel := context.WithTimeout(ctx, setupTimeout) + err = run.setup(setupCtx) + setupCancel() + if err != nil { + return xerrors.Errorf("set up: %w", err) + } + preparedUsers := run.users + + _, _ = fmt.Fprintf(inv.Stderr, "Set up %d users (%d template admins)\n", len(preparedUsers), templateAdminCount) + + dialBarrier := &sync.WaitGroup{} + templateAdminWatchBarrier := &sync.WaitGroup{} + dialBarrier.Add(len(preparedUsers)) + templateAdminWatchBarrier.Add(int(templateAdminCount)) + + // --timeout is the single budget for the measured phase: connecting every + // runner, triggering the notifications, and observing them arrive. Both the + // trigger and every runner derive from it, so there is no per-run timeout + // nested inside it that could expire first. WithCancelCause lets a trigger + // failure end the run immediately with its own error, instead of leaving + // every runner to wait out the budget and report a deadline instead. + cancelCtx, cancelTest := context.WithCancelCause(ctx) + defer cancelTest(nil) + testCtx, testTimeoutCancel := context.WithTimeout(cancelCtx, testTimeout) + defer testTimeoutCancel() + + triggerDone := make(chan struct{}) + go func() { + defer close(triggerDone) + if err := triggerNotifications( + testCtx, + logger, + client, + orgID, + runName, + dialBarrier, + triggerTimes, + ); err != nil { + logger.Error(ctx, "failed to trigger notifications", slog.Error(err)) + cancelTest(xerrors.Errorf("trigger notifications: %w", err)) + } + }() + + // The runners are not harness.Cleanable: this command owns the user + // lifecycle and tears it down in bulk above, so th.Cleanup is never called + // and the cleanup strategy passed here is inert. + th := harness.NewTestHarness(harness.ConcurrentExecutionStrategy{}, harness.LinearExecutionStrategy{}) + + for i, pu := range preparedUsers { + id := strconv.Itoa(i) + name := fmt.Sprintf("notifications-%s", id) + config := notifications.Config{ - User: createusers.Config{ - OrganizationID: me.OrganizationIDs[0], - }, - Roles: []string{}, - NotificationTimeout: notificationTimeout, + PreCreatedUser: pu.user, + SessionToken: pu.sessionToken, + URL: client.URL, + DialHTTPClient: dialHTTPClient, DialTimeout: dialTimeout, DialBarrier: dialBarrier, ReceivingWatchBarrier: templateAdminWatchBarrier, Metrics: metrics, } + if pu.isAdmin { + config.ExpectedNotificationIDs = expectedNotificationIDs + config.SMTPApiURL = smtpAPIURL + config.SMTPRequestTimeout = smtpRequestTimeout + config.SMTPHttpClient = smtpHTTPClient + } if err := config.Validate(); err != nil { return xerrors.Errorf("validate config: %w", err) } - configs = append(configs, config) - } - - go triggerNotifications( - ctx, - logger, - client, - me.OrganizationIDs[0], - dialBarrier, - dialTimeout, - triggerTimes, - ) - th := harness.NewTestHarness(timeoutStrategy.wrapStrategy(harness.ConcurrentExecutionStrategy{}), cleanupStrategy.toStrategy()) - - for i, config := range configs { - id := strconv.Itoa(i) - name := fmt.Sprintf("notifications-%s", id) - // use an independent client for each Runner, so they don't reuse TCP connections. This can lead to - // requests being unbalanced among Coder instances. - runnerClient, err := loadtestutil.DupClientCopyingHeaders(client, BypassHeader) - if err != nil { - return xerrors.Errorf("create runner client: %w", err) - } - var runner harness.Runnable = notifications.NewRunner(runnerClient, config) + var runner harness.Runnable = notifications.NewRunner(config) if tracingEnabled { runner = &runnableTraceWrapper{ tracer: tracer, @@ -216,13 +387,23 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { } _, _ = fmt.Fprintln(inv.Stderr, "Running notification delivery scaletest...") - testCtx, testCancel := timeoutStrategy.toContext(ctx) - defer testCancel() err = th.Run(testCtx) if err != nil { return xerrors.Errorf("run test harness (harness failure, not a test failure): %w", err) } + // Wait for the trigger goroutine before reading results. A runner only needs + // the template delete to have happened to receive its notification, so + // th.Run can return before the trigger records its time. Reading results + // first would then find no trigger time and report zero latencies. + <-triggerDone + + // A trigger failure cancels testCtx with its own cause. Report that instead + // of the N per-runner deadline errors it produces, which point at delivery. + if cause := context.Cause(testCtx); cause != nil && !errors.Is(cause, context.Canceled) && !errors.Is(cause, context.DeadlineExceeded) { + return cause + } + // If the command was interrupted, skip stats. if notifyCtx.Err() != nil { return notifyCtx.Err() @@ -230,7 +411,7 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { res := th.Results() - if err := computeNotificationLatencies(ctx, logger, triggerTimes, res, metrics); err != nil { + if err := computeNotificationLatencies(ctx, logger, metrics, triggerTimes, res); err != nil { return xerrors.Errorf("compute notification latencies: %w", err) } @@ -241,16 +422,6 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { } } - if !noCleanup { - _, _ = fmt.Fprintln(inv.Stderr, "\nCleaning up...") - cleanupCtx, cleanupCancel := cleanupStrategy.toContext(ctx) - defer cleanupCancel() - err = th.Cleanup(cleanupCtx) - if err != nil { - return xerrors.Errorf("cleanup tests: %w", err) - } - } - if res.TotalFail > 0 { return xerrors.New("load test failed, see above for more details") } @@ -264,7 +435,7 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { Flag: "user-count", FlagShorthand: "c", Env: "CODER_SCALETEST_NOTIFICATION_USER_COUNT", - Description: "Required: Total number of users to create.", + Description: "Required: Total number of users to run as. Users are created and deleted by this command.", Value: serpent.Int64Of(&userCount), Required: true, }, @@ -272,15 +443,22 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { Flag: "template-admin-percentage", Env: "CODER_SCALETEST_NOTIFICATION_TEMPLATE_ADMIN_PERCENTAGE", Default: "20.0", - Description: "Percentage of users to assign Template Admin role to (0-100).", + Description: "Percentage of users to assign the Template Admin role to, which decides how many receive the notification under test. Must be greater than 0 and at most 100.", Value: serpent.Float64Of(&templateAdminPercentage), }, { - Flag: "notification-timeout", + Flag: "setup-concurrency", + Env: "CODER_SCALETEST_NOTIFICATION_SETUP_CONCURRENCY", + Default: "10", + Description: "Number of concurrent workers used to set up users before the test and to clean them up afterwards. Bounds how many connections those phases hold regardless of user count. Must be greater than 0.", + Value: serpent.Int64Of(&setupConcurrency), + }, + { + Flag: "timeout", Env: "CODER_SCALETEST_NOTIFICATION_TIMEOUT", - Default: "10m", - Description: "How long to wait for notifications after triggering.", - Value: serpent.DurationOf(¬ificationTimeout), + Default: "30m", + Description: "Overall budget for the measured phase: connecting every runner, triggering the notifications, and waiting for them to arrive. Every runner is canceled when it expires. Must be greater than 0; unlike the shared scaletest timeout flags, 0 is not accepted as unlimited, which is why this uses its own environment variable.", + Value: serpent.DurationOf(&testTimeout), }, { Flag: "smtp-request-timeout", @@ -291,9 +469,9 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { }, { Flag: "dial-timeout", - Env: "CODER_SCALETEST_DIAL_TIMEOUT", + Env: "CODER_SCALETEST_NOTIFICATION_DIAL_TIMEOUT", Default: "10m", - Description: "Timeout for dialing the notification websocket endpoint.", + Description: "Timeout for dialing the notification websocket endpoint. Must be greater than 0 and less than --timeout; like the other timeouts here it does not accept 0 as unlimited, which is why it uses its own environment variable.", Value: serpent.DurationOf(&dialTimeout), }, { @@ -302,6 +480,20 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { Description: "Do not clean up resources after the test completes.", Value: serpent.BoolOf(&noCleanup), }, + { + Flag: "setup-timeout", + Env: "CODER_SCALETEST_NOTIFICATION_SETUP_TIMEOUT", + Default: defaultPhaseTimeout.String(), + Description: "Timeout for the setup phase, covering user creation or selection and clearing stale trigger templates. Defaults to the same value as --cleanup-timeout, which undoes the same work. Must be greater than 0.", + Value: serpent.DurationOf(&setupTimeout), + }, + { + Flag: "cleanup-timeout", + Env: "CODER_SCALETEST_NOTIFICATION_CLEANUP_TIMEOUT", + Default: defaultPhaseTimeout.String(), + Description: "Timeout for the whole cleanup phase, covering users and the trigger template. Defaults to the same value as --setup-timeout, which creates the same work. Must be greater than 0; unlike the shared scaletest cleanup flags, 0 is not accepted as unlimited, which is why this uses its own environment variable.", + Value: serpent.DurationOf(&cleanupTimeout), + }, { Flag: "smtp-api-url", Env: "CODER_SCALETEST_SMTP_API_URL", @@ -311,8 +503,6 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { } tracingFlags.attach(&cmd.Options) - timeoutStrategy.attach(&cmd.Options) - cleanupStrategy.attach(&cmd.Options) output.attach(&cmd.Options) prometheusFlags.attach(&cmd.Options) return cmd @@ -321,9 +511,9 @@ func (r *RootCmd) scaletestNotifications() *serpent.Command { func computeNotificationLatencies( ctx context.Context, logger slog.Logger, + metrics *notifications.Metrics, expectedNotifications map[uuid.UUID]chan time.Time, results harness.Results, - metrics *notifications.Metrics, ) error { triggerTimes := make(map[uuid.UUID]time.Time) for notificationID, triggerTimeChan := range expectedNotifications { @@ -394,23 +584,31 @@ func computeNotificationLatencies( return nil } -// triggerNotifications waits for all test users to connect, -// then creates and deletes a test template to trigger notification events for testing. +// triggerNotifications waits for all test users to connect, then creates and +// deletes a template to trigger notification events for testing. The template +// name carries a per-run suffix so a template stranded by a hard kill cannot make +// later runs fail on a name conflict, and so two runs do not collide. +// +// Any error is returned rather than only logged: every runner is parked waiting +// for a notification that can now never arrive, so the caller must fail the run +// instead of letting the whole fleet wait out the budget. func triggerNotifications( ctx context.Context, logger slog.Logger, client *codersdk.Client, orgID uuid.UUID, + templateName string, dialBarrier *sync.WaitGroup, - dialTimeout time.Duration, expectedNotifications map[uuid.UUID]chan time.Time, -) { - logger.Info(ctx, "waiting for all users to connect") +) error { + triggered := newTriggerRecorder(expectedNotifications) - // Wait for all users to connect - waitCtx, cancel := context.WithTimeout(ctx, dialTimeout+30*time.Second) - defer cancel() + logger.Info(ctx, "waiting for all users to connect") + // Wait for every runner to connect. Bounded by ctx alone: it already carries + // the overall budget, and --dial-timeout is validated to leave headroom inside + // it, so a separately derived window would only add a second deadline that can + // expire while runners are still legitimately dialing. done := make(chan struct{}) go func() { dialBarrier.Wait() @@ -420,13 +618,8 @@ func triggerNotifications( select { case <-done: logger.Info(ctx, "all users connected") - case <-waitCtx.Done(): - if waitCtx.Err() == context.DeadlineExceeded { - logger.Error(ctx, "timeout waiting for users to connect") - } else { - logger.Info(ctx, "context canceled while waiting for users") - } - return + case <-ctx.Done(): + return xerrors.Errorf("wait for users to connect: %w", ctx.Err()) } logger.Info(ctx, "creating test template to test notifications") @@ -434,8 +627,7 @@ func triggerNotifications( // Upload empty template file. file, err := client.Upload(ctx, codersdk.ContentTypeTar, bytes.NewReader([]byte{})) if err != nil { - logger.Error(ctx, "upload test template", slog.Error(err)) - return + return xerrors.Errorf("upload test template: %w", err) } logger.Info(ctx, "test template uploaded", slog.F("file_id", file.ID)) @@ -446,32 +638,402 @@ func triggerNotifications( Provisioner: codersdk.ProvisionerTypeEcho, }) if err != nil { - logger.Error(ctx, "create test template version", slog.Error(err)) - return + return xerrors.Errorf("create test template version: %w", err) } logger.Info(ctx, "test template version created", slog.F("template_version_id", version.ID)) // Create template. testTemplate, err := client.CreateTemplate(ctx, orgID, codersdk.CreateTemplateRequest{ - Name: "scaletest-test-template", - Description: "scaletest-test-template", + Name: templateName, + Description: templateName, VersionID: version.ID, }) if err != nil { - logger.Error(ctx, "create test template", slog.Error(err)) - return + return xerrors.Errorf("create test template: %w", err) } logger.Info(ctx, "test template created", slog.F("template_id", testTemplate.ID)) // Delete template to trigger notification. - err = client.DeleteTemplate(ctx, testTemplate.ID) - if err != nil { - logger.Error(ctx, "delete test template", slog.Error(err)) - return + if err := client.DeleteTemplate(ctx, testTemplate.ID); err != nil { + return xerrors.Errorf("delete test template: %w", err) } logger.Info(ctx, "test template deleted", slog.F("template_id", testTemplate.ID)) - // Record expected notification. - expectedNotifications[notificationsLib.TemplateTemplateDeleted] <- time.Now() - close(expectedNotifications[notificationsLib.TemplateTemplateDeleted]) + // Deleting the template is the action that produces TemplateTemplateDeleted, so + // record the trigger time against that notification specifically rather than + // against every notification the runners happen to expect. + if err := triggered.record(notificationsLib.TemplateTemplateDeleted); err != nil { + return err + } + + // Every expected notification must have been produced by one of the actions + // above. Without this check, adding an ID to the expected set without adding the + // action that causes it leaves every runner waiting for a notification that can + // never arrive, until the budget expires. + return triggered.verifyAll() +} + +// triggerRecorder records when the action producing each expected notification +// completed, and reports an expected notification that no action produced. +type triggerRecorder struct { + expected map[uuid.UUID]chan time.Time + done map[uuid.UUID]struct{} +} + +func newTriggerRecorder(expected map[uuid.UUID]chan time.Time) *triggerRecorder { + return &triggerRecorder{ + expected: expected, + done: make(map[uuid.UUID]struct{}, len(expected)), + } +} + +// record stores the current time as the trigger time for notificationID. The +// channel is buffered and closed after the send, so the read in +// computeNotificationLatencies finds the value whenever it runs. +func (t *triggerRecorder) record(notificationID uuid.UUID) error { + ch, ok := t.expected[notificationID] + if !ok { + return xerrors.Errorf("triggered notification %q that no runner is waiting for", notificationID) + } + if _, ok := t.done[notificationID]; ok { + return xerrors.Errorf("notification %q triggered more than once", notificationID) + } + ch <- time.Now() + close(ch) + t.done[notificationID] = struct{}{} + return nil +} + +// verifyAll reports any expected notification that no action triggered. +func (t *triggerRecorder) verifyAll() error { + var errs error + for id := range t.expected { + if _, ok := t.done[id]; !ok { + errs = errors.Join(errs, xerrors.Errorf( + "no action triggers expected notification %q, so no runner can receive it", id)) + } + } + return errs +} + +// scaletestRun owns every resource this run creates or changes on the deployment: +// the users the runners connect as, and the template whose deletion triggers the +// notification under test. +// +// Setup and cleanup are each a single entry point covering both resources, so they +// cannot drift apart in timeout, concurrency, progress reporting, or interrupt +// handling. Cleaning the two up separately previously gave each its own full +// --cleanup-timeout, so the worst case was twice the budget the operator asked for. +type scaletestRun struct { + logger slog.Logger + stderr io.Writer + metrics *notifications.Metrics + + // client performs template operations as the calling admin. setupClient has a + // bounded connection pool for the per-user work. + client *codersdk.Client + setupClient *codersdk.Client + + orgID uuid.UUID + concurrency int + noCleanup bool + + userCount int + adminCount int + templateName string + + // users is filled by setup and read by cleanup, so a failure part-way through + // setup still tears down what it managed to create. + users []preparedUser +} + +// setup clears any stale trigger template and makes the users the runners connect +// as. It is bounded by the caller's context and runs at most concurrency requests +// at a time. +func (r *scaletestRun) setup(ctx context.Context) error { + // A template stranded by an earlier killed run would accumulate. Per-run names + // keep one from breaking this run, so this only stops the pile-up and failing to + // sweep is not fatal. Skipped under --no-cleanup, which promises to delete + // nothing. + if !r.noCleanup { + if err := r.sweepStaleTemplates(ctx); err != nil { + r.logger.Warn(ctx, "failed to sweep stale trigger templates", slog.Error(err)) + } + } + + return r.createUsers(ctx) +} + +// cleanup undoes everything setup and the trigger created, best-effort, and +// returns the joined errors. Users come first: an elevated role and a live token +// on an account matter more than a leftover template. +func (r *scaletestRun) cleanup(ctx context.Context) error { + errs := r.cleanupUsers(ctx) + + // Delete by name rather than by an ID recorded when the create returned. A + // create can commit on the server while the client sees an error, and the name + // is chosen before the request is made, so the lookup finds the template either + // way. It is also already gone on the happy path, where the trigger deleted it + // to fire the notification. + if err := r.deleteTemplateByName(ctx, r.templateName); err != nil { + errs = errors.Join(errs, xerrors.Errorf("clean up trigger template: %w", err)) + } + return errs +} + +func (r *scaletestRun) createUsers(ctx context.Context) error { + r.users = make([]preparedUser, r.userCount) + for i := range r.users { + r.users[i] = preparedUser{ + origin: originCreated, + isAdmin: i < r.adminCount, + id: strconv.Itoa(i), + } + } + + // A separate client because RunReturningUser turns on body logging for whatever + // client it is handed, which would follow the shared setup client into cleanup. + createClient, err := loadtestutil.DupClientConfiguringTransport(r.client, BypassHeader, boundPool(r.concurrency)) + if err != nil { + return xerrors.Errorf("create user-creation client: %w", err) + } + + _, _ = fmt.Fprintf(r.stderr, "Creating %d users (%d template admins) with %d concurrent workers...\n", + r.userCount, r.adminCount, r.concurrency) + return forEachUser(ctx, r.users, r.concurrency, func(ctx context.Context, pu *preparedUser) error { + if err := createAndLoginUser(ctx, createClient, r.orgID, pu); err != nil { + r.metrics.AddError("create_user") + return xerrors.Errorf("create user %q: %w", pu.id, err) + } + return nil + }) +} + +// cleanupUsers deletes the users this run created. +func (r *scaletestRun) cleanupUsers(ctx context.Context) error { + if len(r.users) == 0 { + return nil + } + return forEachUserBestEffort(ctx, r.users, r.concurrency, func(ctx context.Context, pu *preparedUser) error { + return deleteUser(ctx, r.metrics, r.setupClient, pu) + }) +} + +// deleteTemplateByName deletes the named template if it still exists. +func (r *scaletestRun) deleteTemplateByName(ctx context.Context, name string) error { + tpl, err := r.client.TemplateByName(ctx, r.orgID, name) + if err != nil { + if sdkErr, ok := errors.AsType[*codersdk.Error](err); ok && sdkErr.StatusCode() == http.StatusNotFound { + // Never created, or already deleted by the trigger. + return nil + } + return xerrors.Errorf("look up template %q: %w", name, err) + } + if err := r.client.DeleteTemplate(ctx, tpl.ID); err != nil { + return xerrors.Errorf("delete template %q: %w", name, err) + } + return nil +} + +// sweepStaleTemplates deletes trigger templates left behind by earlier runs that +// were killed between creating and deleting one. Per-run names keep a stranded +// template from breaking this run, so this only stops them accumulating. +// +// Every trigger template except this run's own is fair game. A concurrent run's +// in-flight template is indistinguishable from a stranded one, which is one of the +// reasons only one instance of this command may run against a deployment at a +// time. +func (r *scaletestRun) sweepStaleTemplates(ctx context.Context) error { + // Filter server-side rather than listing every template in the deployment. The + // filter is a substring match, so the prefix check below still decides. + templates, err := r.client.Templates(ctx, codersdk.TemplateFilter{ + OrganizationID: r.orgID, + FuzzyName: notificationsPrefix, + }) + if err != nil { + return xerrors.Errorf("list templates: %w", err) + } + var errs error + for _, tpl := range templates { + if tpl.Name == r.templateName || !strings.HasPrefix(tpl.Name, notificationsPrefix) { + continue + } + r.logger.Info(ctx, "deleting stale trigger template left by an earlier run", + slog.F("template_id", tpl.ID), slog.F("template_name", tpl.Name)) + if err := r.client.DeleteTemplate(ctx, tpl.ID); err != nil { + errs = errors.Join(errs, xerrors.Errorf("delete stale template %q: %w", tpl.Name, err)) + } + } + return errs +} + +// boundPool returns a transport configuration that caps the connection pool at +// limit and keeps those connections warm, so a phase making limit concurrent +// requests reuses connections instead of re-dialing per request. +func boundPool(limit int) func(*http.Transport) { + return func(t *http.Transport) { + t.MaxIdleConns = limit + t.MaxIdleConnsPerHost = limit + t.MaxConnsPerHost = limit + t.IdleConnTimeout = scaletestIdleConnTimeout + } +} + +// forEachUser runs fn once per user, at most limit at a time. The first error +// cancels the shared context so in-flight and pending calls stop early, and that +// error is returned. limit must be positive, which the CLI validates. +func forEachUser(ctx context.Context, users []preparedUser, limit int, fn func(context.Context, *preparedUser) error) error { + eg, egCtx := errgroup.WithContext(ctx) + eg.SetLimit(limit) + for i := range users { + pu := &users[i] + eg.Go(func() error { + if err := egCtx.Err(); err != nil { + return err + } + return fn(egCtx, pu) + }) + } + return eg.Wait() +} + +// forEachUserBestEffort runs fn once per user, at most limit at a time. Every call +// runs to completion even when one fails, and all returned errors are joined. +// Cleanup uses this so a single failure cannot abandon the remaining users. +func forEachUserBestEffort(ctx context.Context, users []preparedUser, limit int, fn func(context.Context, *preparedUser) error) error { + var ( + mu sync.Mutex + errs error + ) + var eg errgroup.Group + eg.SetLimit(limit) + for i := range users { + pu := &users[i] + eg.Go(func() error { + if err := fn(ctx, pu); err != nil { + mu.Lock() + errs = errors.Join(errs, err) + mu.Unlock() + return nil + } + return nil + }) + } + _ = eg.Wait() + return errs +} + +// preparedUser is a user made ready to connect for the run, authenticated with +// a session token, plus the metadata cleanup needs to delete it. A new user is +// created and logged in, and cleanup deletes it by ID. +// +// userOrigin records where a prepared user came from, which decides what cleanup +// is allowed to do to it. A user this run did not create must never be deleted, +// and the sink enforces that rather than trusting its callers. +type userOrigin int + +const ( + // originCreated means this run created the user and must delete it. + originCreated userOrigin = iota +) + +type preparedUser struct { + origin userOrigin + user codersdk.User + sessionToken string + // isAdmin marks a user designated as a template admin for the run. + isAdmin bool + + // id identifies the user within the run; the create runner generates the + // username and email from it. + id string +} + +// createAndLoginUser creates the user in orgID and logs in to obtain a session +// token, then promotes it to template admin when designated, filling in the +// mutable fields of pu. pu.user is set as soon as the user exists so a later +// failure still lets cleanup delete it. +// +// client must not be the shared setup client: RunReturningUser calls SetLogger and +// SetLogBodies on whatever it is given, which would make the setup client copy +// every request and response body for the rest of setup and all of cleanup. +func createAndLoginUser(ctx context.Context, client *codersdk.Client, orgID uuid.UUID, pu *preparedUser) error { + // Reuse createusers.Runner so the create+login sequence lives in one place. It + // generates the username and email from the id when Config leaves them empty. + runner := createusers.NewRunner(client, createusers.Config{OrganizationID: orgID}) + created, err := runner.RunReturningUser(ctx, pu.id, io.Discard) + // Capture the user even on failure: if creation succeeded but a later step + // failed, cleanup still needs the ID to delete it. + pu.user = runner.User() + if err != nil { + // The create may have completed server-side while the client gave up, which + // leaves no ID for cleanup to delete. Look the user up by the name the runner + // generated so it is not orphaned. Best-effort on a context that is likely + // already done, so failure here only loses what was already lost. + if pu.user.ID == uuid.Nil { + if found, ferr := findUserByScaletestID(ctx, client, pu.id); ferr == nil { + pu.user = found + } + } + return xerrors.Errorf("create and login user: %w", err) + } + pu.sessionToken = created.SessionToken + + // The create runner does not assign roles. + if pu.isAdmin { + if _, err := client.UpdateUserRoles(ctx, pu.user.ID.String(), codersdk.UpdateRoles{Roles: []string{codersdk.RoleTemplateAdmin}}); err != nil { + return xerrors.Errorf("assign template admin role: %w", err) + } + } + + return nil +} + +// findUserByScaletestID looks for a user this run created, identified by the id +// suffix the create runner puts in the generated username. Used to recover the ID +// of a user whose creation completed after the client stopped waiting. +// +// The suffix is not unique across runs, so this can match a user left behind by an +// earlier run with the same index. That is acceptable only because a single +// operator runs this command against a deployment at a time, and because both +// candidates are disposable scaletest users that "coder exp scaletest cleanup" +// reclaims either way. +func findUserByScaletestID(ctx context.Context, client *codersdk.Client, id string) (codersdk.User, error) { + resp, err := client.Users(ctx, codersdk.UsersRequest{ + SearchQuery: loadtestutil.ScaleTestPrefix, + Pagination: codersdk.Pagination{Limit: 1000}, + }) + if err != nil { + return codersdk.User{}, xerrors.Errorf("list users: %w", err) + } + suffix := "-" + id + for _, u := range resp.Users { + if strings.HasPrefix(u.Username, loadtestutil.ScaleTestPrefix+"-") && strings.HasSuffix(u.Username, suffix) { + return u, nil + } + } + return codersdk.User{}, xerrors.Errorf("no user found for scaletest id %q", id) +} + +// deleteUser deletes the user by ID, best-effort. Callers must pass only users +// this test created; it must never see reuse-path users, whose accounts belong to +// the deployment. +func deleteUser(ctx context.Context, metrics *notifications.Metrics, client *codersdk.Client, u *preparedUser) error { + // Refuse here rather than trusting the call site. Deleting a reused account is + // unrecoverable, so the check belongs where the damage would be done. + if u.origin != originCreated { + return xerrors.Errorf("refusing to delete user %q that this run did not create", u.user.ID) + } + if u.user.ID == uuid.Nil { + return nil + } + if err := client.DeleteUser(ctx, u.user.ID); err != nil { + metrics.AddError("delete_user") + return xerrors.Errorf("delete user %q: %w", u.user.ID, err) + } + return nil +} + +func userHasRole(user codersdk.User, role string) bool { + return slices.ContainsFunc(user.Roles, func(r codersdk.SlimRole) bool { return r.Name == role }) } diff --git a/cli/exp_scaletest_notifications_internal_test.go b/cli/exp_scaletest_notifications_internal_test.go new file mode 100644 index 00000000000..b9131975fc3 --- /dev/null +++ b/cli/exp_scaletest_notifications_internal_test.go @@ -0,0 +1,388 @@ +//go:build !slim + +package cli + +import ( + "context" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + "golang.org/x/xerrors" + + "github.com/coder/coder/v2/coderd/coderdtest" + "github.com/coder/coder/v2/codersdk" + "github.com/coder/coder/v2/scaletest/loadtestutil" + "github.com/coder/coder/v2/scaletest/notifications" + "github.com/coder/coder/v2/testutil" +) + +func TestCreateAndDeleteUsers(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitLong) + client := coderdtest.New(t, nil) + firstUser := coderdtest.CreateFirstUser(t, client) + metrics := notifications.NewMetrics(prometheus.NewRegistry()) + + const userCount = 4 + const adminCount = 2 + candidates := make([]preparedUser, userCount) + for i := range candidates { + candidates[i] = preparedUser{origin: originCreated, isAdmin: i < adminCount, id: strconv.Itoa(i)} + } + + err := forEachUser(ctx, candidates, 2, func(ctx context.Context, pu *preparedUser) error { + return createAndLoginUser(ctx, client, firstUser.OrganizationID, pu) + }) + require.NoError(t, err) + + users, err := client.Users(ctx, codersdk.UsersRequest{}) + require.NoError(t, err) + require.Len(t, users.Users, userCount+1) + + for i, pu := range candidates { + require.NotEqual(t, uuid.Nil, pu.user.ID) + require.NotEmpty(t, pu.sessionToken) + // Created users must be reusable by a later run. + require.True(t, loadtestutil.IsScaleTestUser(pu.user.Username, pu.user.Email)) + + userClient := codersdk.New(client.URL, codersdk.WithSessionToken(pu.sessionToken)) + me, err := userClient.User(ctx, codersdk.Me) + require.NoError(t, err) + require.Equal(t, pu.user.ID, me.ID) + + got, err := client.User(ctx, pu.user.ID.String()) + require.NoError(t, err) + require.Equal(t, i < adminCount, userHasRole(got, codersdk.RoleTemplateAdmin)) + } + + err = forEachUserBestEffort(ctx, candidates, 2, func(ctx context.Context, pu *preparedUser) error { + return deleteUser(ctx, metrics, client, pu) + }) + require.NoError(t, err) + + users, err = client.Users(ctx, codersdk.UsersRequest{}) + require.NoError(t, err) + require.Len(t, users.Users, 1) + require.Equal(t, firstUser.UserID, users.Users[0].ID) +} + +// failingPathClient returns a client whose requests reach the real deployment +// except those to failPath, which fail. Used to make one step of a multi-step +// helper fail after earlier steps have already taken effect server-side. +func failingPathClient(t *testing.T, target *codersdk.Client, failPath string) *codersdk.Client { + t.Helper() + + proxy := httputil.NewSingleHostReverseProxy(target.URL) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == failPath { + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(`{"message":"injected failure"}`)) + return + } + proxy.ServeHTTP(w, r) + })) + t.Cleanup(srv.Close) + + proxyURL, err := url.Parse(srv.URL) + require.NoError(t, err) + client := codersdk.New(proxyURL, codersdk.WithSessionToken(target.SessionToken())) + return client +} + +// TestCreateAndLoginUserCapturesPartialUser covers why createAndLoginUser assigns +// pu.user before checking the error: creation can succeed while a later step +// fails, and cleanup needs the ID to reclaim the user. +func TestCreateAndLoginUserCapturesPartialUser(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitLong) + client := coderdtest.New(t, nil) + firstUser := coderdtest.CreateFirstUser(t, client) + metrics := notifications.NewMetrics(prometheus.NewRegistry()) + + // Creation succeeds; the login that follows it fails. + failingClient := failingPathClient(t, client, "/api/v2/users/login") + + pu := &preparedUser{origin: originCreated, id: "partial"} + err := createAndLoginUser(ctx, failingClient, firstUser.OrganizationID, pu) + require.Error(t, err) + require.Empty(t, pu.sessionToken, "login did not complete") + + // The mechanism under test: the created user survives the failure, so cleanup + // can still find it. + require.NotEqual(t, uuid.Nil, pu.user.ID) + require.True(t, loadtestutil.IsScaleTestUser(pu.user.Username, pu.user.Email)) + + users, err := client.Users(ctx, codersdk.UsersRequest{}) + require.NoError(t, err) + require.Len(t, users.Users, 2, "the user was really created") + + require.NoError(t, deleteUser(ctx, metrics, client, pu)) + + users, err = client.Users(ctx, codersdk.UsersRequest{}) + require.NoError(t, err) + require.Len(t, users.Users, 1, "the partially created user is reclaimed") + + // A candidate with no ID must be a no-op rather than an error. + require.NoError(t, deleteUser(ctx, metrics, client, &preparedUser{origin: originCreated, id: "never"})) +} + +func makeTestUsers(n int) []preparedUser { + users := make([]preparedUser, n) + for i := range users { + users[i].user = codersdk.User{ + ReducedUser: codersdk.ReducedUser{ + MinimalUser: codersdk.MinimalUser{ID: uuid.New()}, + }, + } + } + return users +} + +func TestForEachUser(t *testing.T) { + t.Parallel() + + t.Run("CoversEveryUserOnceWithinLimit", func(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitShort) + users := makeTestUsers(10) + const limit = 3 + var ( + mu sync.Mutex + seen = map[uuid.UUID]int{} + inFlight atomic.Int64 + maxSeen atomic.Int64 + ) + // Block until the limit is reached before returning, so a raised limit really + // does show up as more calls in flight. Without this every call can finish + // before the next starts and the assertion holds for any limit. + atLimit := make(chan struct{}) + var once sync.Once + err := forEachUser(ctx, users, limit, func(_ context.Context, pu *preparedUser) error { + n := inFlight.Add(1) + if n == int64(limit) { + once.Do(func() { close(atLimit) }) + } + <-atLimit + for { + old := maxSeen.Load() + if n <= old || maxSeen.CompareAndSwap(old, n) { + break + } + } + defer inFlight.Add(-1) + + mu.Lock() + seen[pu.user.ID]++ + mu.Unlock() + return nil + }) + require.NoError(t, err) + require.Len(t, seen, 10) + for _, count := range seen { + require.Equal(t, 1, count, "each user is processed exactly once") + } + require.LessOrEqual(t, maxSeen.Load(), int64(limit), "never exceeds the concurrency limit") + }) + + t.Run("EmptySliceSkipsFn", func(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitShort) + called := false + err := forEachUser(ctx, nil, 3, func(context.Context, *preparedUser) error { + called = true + return nil + }) + require.NoError(t, err) + require.False(t, called) + }) + + t.Run("CancelsSiblingsOnError", func(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitShort) + users := makeTestUsers(2) + sentinel := xerrors.New("boom") + var siblingCanceled atomic.Bool + siblingRunning := make(chan struct{}) + // Both users run concurrently under the limit. The second only fails once the + // first is known to be inside fn, so the first is guaranteed to observe the + // cancellation instead of being skipped before it starts. + err := forEachUser(ctx, users, 2, func(ctx context.Context, pu *preparedUser) error { + if pu.user.ID == users[0].user.ID { + close(siblingRunning) + select { + case <-ctx.Done(): + siblingCanceled.Store(true) + return ctx.Err() + case <-time.After(testutil.WaitShort): + return nil + } + } + <-siblingRunning + return sentinel + }) + require.ErrorIs(t, err, sentinel) + require.True(t, siblingCanceled.Load()) + }) +} + +func TestForEachUserBestEffort(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitShort) + users := makeTestUsers(4) + errA := xerrors.New("err-a") + errB := xerrors.New("err-b") + var ( + mu sync.Mutex + processed int + observedCancel bool + ) + // Every user is processed and its error joined; a failure never cancels the + // others, which is what stops one failed demotion abandoning the rest. + err := forEachUserBestEffort(ctx, users, 2, func(ctx context.Context, pu *preparedUser) error { + mu.Lock() + processed++ + if ctx.Err() != nil { + observedCancel = true + } + mu.Unlock() + if pu.user.ID == users[0].user.ID { + return errA + } + return errB + }) + require.Equal(t, 4, processed) + require.False(t, observedCancel, "a failing call must not cancel siblings") + require.ErrorIs(t, err, errA) + require.ErrorIs(t, err, errB) +} + +// TestTriggerRecorder covers the coupling between the notifications the runners +// wait for and the actions the trigger actually performs. Drift between the two +// used to be silent: a runner waited for a notification nothing sent, and the run +// burned its whole budget before failing with delivery errors. +func TestTriggerRecorder(t *testing.T) { + t.Parallel() + + expectedID := uuid.New() + + t.Run("RecordsAndVerifies", func(t *testing.T) { + t.Parallel() + + ch := make(chan time.Time, 1) + rec := newTriggerRecorder(map[uuid.UUID]chan time.Time{expectedID: ch}) + + before := time.Now() + require.NoError(t, rec.record(expectedID)) + require.NoError(t, rec.verifyAll()) + + // The trigger time is readable and the channel closed, so the latency + // computation cannot miss it. + got, ok := <-ch + require.True(t, ok) + require.False(t, got.Before(before)) + _, stillOpen := <-ch + require.False(t, stillOpen, "channel is closed after recording") + }) + + t.Run("ExpectedNotificationWithNoAction", func(t *testing.T) { + t.Parallel() + + // A second expected notification that no action triggers must be reported, + // not silently waited on by the runners. + unsentID := uuid.New() + rec := newTriggerRecorder(map[uuid.UUID]chan time.Time{ + expectedID: make(chan time.Time, 1), + unsentID: make(chan time.Time, 1), + }) + require.NoError(t, rec.record(expectedID)) + + err := rec.verifyAll() + require.ErrorContains(t, err, unsentID.String()) + require.ErrorContains(t, err, "no action triggers expected notification") + }) + + t.Run("TriggeredNotificationNobodyExpects", func(t *testing.T) { + t.Parallel() + + rec := newTriggerRecorder(map[uuid.UUID]chan time.Time{}) + err := rec.record(expectedID) + require.ErrorContains(t, err, "no runner is waiting for") + }) + + t.Run("DoubleRecordIsRejected", func(t *testing.T) { + t.Parallel() + + rec := newTriggerRecorder(map[uuid.UUID]chan time.Time{expectedID: make(chan time.Time, 1)}) + require.NoError(t, rec.record(expectedID)) + // Recording twice would panic on the closed channel, so it is rejected. + require.ErrorContains(t, rec.record(expectedID), "triggered more than once") + }) +} + +// createEmptyTemplate creates a template with the given name so a test can assert +// what the sweep does and does not delete. +func createEmptyTemplate(ctx context.Context, t *testing.T, client *codersdk.Client, orgID uuid.UUID, name string) codersdk.Template { + t.Helper() + + version := coderdtest.CreateTemplateVersion(t, client, orgID, nil) + coderdtest.AwaitTemplateVersionJobCompleted(t, client, version.ID) + return coderdtest.CreateTemplate(t, client, orgID, version.ID, func(req *codersdk.CreateTemplateRequest) { + req.Name = name + }) +} + +// TestSweepStaleTemplates covers the sweep that stops trigger templates +// accumulating when runs are killed between creating and deleting one. Its guards +// decide what gets deleted from a live deployment, so each one is exercised here. +func TestSweepStaleTemplates(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitLong) + client := coderdtest.New(t, &coderdtest.Options{IncludeProvisionerDaemon: true}) + firstUser := coderdtest.CreateFirstUser(t, client) + + run := &scaletestRun{ + logger: testutil.Logger(t), + client: client, + orgID: firstUser.OrganizationID, + templateName: notificationsPrefix + "thisrun", + } + + stale := createEmptyTemplate(ctx, t, client, firstUser.OrganizationID, notificationsPrefix+"earlier") + mine := createEmptyTemplate(ctx, t, client, firstUser.OrganizationID, run.templateName) + unrelated := createEmptyTemplate(ctx, t, client, firstUser.OrganizationID, "production-template") + // Contains the prefix but does not start with it. The server-side filter is a + // substring match, so this comes back from the query and only the client-side + // prefix check keeps it alive. + lookalike := createEmptyTemplate(ctx, t, client, firstUser.OrganizationID, "x-"+notificationsPrefix+"c") + + require.NoError(t, run.sweepStaleTemplates(ctx)) + + remaining, err := client.TemplatesByOrganization(ctx, firstUser.OrganizationID) + require.NoError(t, err) + + names := make([]string, 0, len(remaining)) + for _, tpl := range remaining { + names = append(names, tpl.Name) + } + require.NotContains(t, names, stale.Name, "a template left by an earlier run is deleted") + require.Contains(t, names, mine.Name, "this run's own template is kept") + require.Contains(t, names, unrelated.Name, "templates without the prefix are never touched") + require.Contains(t, names, lookalike.Name, "a name containing the prefix is not a name starting with it") +} diff --git a/cli/exp_scaletest_test.go b/cli/exp_scaletest_test.go index 98d2071ad0a..28c7f0add80 100644 --- a/cli/exp_scaletest_test.go +++ b/cli/exp_scaletest_test.go @@ -10,6 +10,7 @@ import ( "cdr.dev/slog/v3/sloggers/slogtest" "github.com/coder/coder/v2/cli/clitest" "github.com/coder/coder/v2/coderd/coderdtest" + "github.com/coder/coder/v2/codersdk" "github.com/coder/coder/v2/testutil" ) @@ -254,3 +255,110 @@ func TestScaleTestDashboard(t *testing.T) { require.ErrorContains(t, err, "invalid target users \"0:0\": start and end cannot be equal") }) } + +// TestScaleTestNotifications_ValidatesArgs checks the degenerate configurations the +// command must reject before doing any work, each of which would otherwise produce +// a run that measures nothing and still exits 0. +func TestScaleTestNotifications_ValidatesArgs(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + args []string + errorContains string + }{ + { + name: "ZeroTemplateAdminPercentage", + args: []string{"--user-count", "1", "--template-admin-percentage", "0"}, + errorContains: "--template-admin-percentage must be greater than 0", + }, + { + name: "ZeroSetupConcurrency", + args: []string{"--user-count", "1", "--setup-concurrency", "0"}, + errorContains: "--setup-concurrency must be greater than 0", + }, + { + name: "ZeroTimeout", + args: []string{"--user-count", "1", "--timeout", "0"}, + errorContains: "--timeout must be greater than 0", + }, + { + name: "ZeroCleanupTimeout", + args: []string{"--user-count", "1", "--cleanup-timeout", "0"}, + errorContains: "--cleanup-timeout must be greater than 0", + }, + { + name: "ZeroSetupTimeout", + args: []string{"--user-count", "1", "--setup-timeout", "0"}, + errorContains: "--setup-timeout must be greater than 0", + }, + { + // The connect phase must leave room to trigger and observe, or the run + // measures nothing. + name: "DialTimeoutNotLessThanTimeout", + args: []string{"--user-count", "1", "--timeout", "30m", "--dial-timeout", "30m"}, + errorContains: "--dial-timeout (30m0s) must be less than --timeout (30m0s)", + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitShort) + log := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}) + client := coderdtest.New(t, &coderdtest.Options{Logger: &log}) + _ = coderdtest.CreateFirstUser(t, client) + + args := append([]string{"exp", "scaletest", "notifications"}, tc.args...) + args = append(args, + "--scaletest-prometheus-address", "127.0.0.1:0", + "--scaletest-prometheus-wait", "0s", + ) + inv, root := clitest.New(t, args...) + clitest.SetupConfig(t, client, root) + err := inv.WithContext(ctx).Run() + require.ErrorContains(t, err, tc.errorContains) + }) + } +} + +// TestScaleTestNotifications_CleanupRunsAfterFailure checks the guarantee that the +// round 1 review found broken: users created during setup are cleaned up even when +// the run fails after setup, not only on the fully successful path. +func TestScaleTestNotifications_CleanupRunsAfterFailure(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitLong) + log := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}) + client := coderdtest.New(t, &coderdtest.Options{Logger: &log}) + firstUser := coderdtest.CreateFirstUser(t, client) + + // A dial timeout this short guarantees the run fails after setup created its + // users, which is the window the cleanup must still cover. + inv, root := clitest.New(t, "exp", "scaletest", "notifications", + "--user-count", "2", + "--template-admin-percentage", "50", + "--timeout", "5s", + "--dial-timeout", "1s", + "--scaletest-prometheus-address", "127.0.0.1:0", + "--scaletest-prometheus-wait", "0s", + ) + // The command requires an admin: it lists users, updates roles, mints tokens, + // and creates templates. + //nolint:gocritic // This scaletest command must run as an admin. + clitest.SetupConfig(t, client, root) + err := inv.WithContext(ctx).Run() + require.Error(t, err, "the run must fail") + + // Every user the run created must be gone, leaving only the owner. + users, err := client.Users(ctx, codersdk.UsersRequest{}) + require.NoError(t, err) + require.Len(t, users.Users, 1, "created users must be cleaned up after a post-setup failure") + require.Equal(t, firstUser.UserID, users.Users[0].ID) + + // The trigger template must not be left behind to poison later runs. + templates, err := client.TemplatesByOrganization(ctx, firstUser.OrganizationID) + require.NoError(t, err) + for _, tpl := range templates { + require.NotContains(t, tpl.Name, "notifications-", "trigger template must be cleaned up") + } +} diff --git a/scaletest/createusers/run.go b/scaletest/createusers/run.go index 78f648f1bc0..e2d5a9e22e4 100644 --- a/scaletest/createusers/run.go +++ b/scaletest/createusers/run.go @@ -99,6 +99,13 @@ func (r *Runner) RunReturningUser(ctx context.Context, id string, logs io.Writer return User{User: user, SessionToken: loginRes.SessionToken}, nil } +// User returns the user created by RunReturningUser, if any. It is set as soon +// as creation succeeds, so callers can still identify a created user for cleanup +// when a later step (such as login) fails. +func (r *Runner) User() codersdk.User { + return r.user +} + func (r *Runner) Cleanup(ctx context.Context, _ string, logs io.Writer) error { if r.user.ID != uuid.Nil { err := r.client.DeleteUser(ctx, r.user.ID) diff --git a/scaletest/loadtestutil/client.go b/scaletest/loadtestutil/client.go index 144b9900898..076b750b0fc 100644 --- a/scaletest/loadtestutil/client.go +++ b/scaletest/loadtestutil/client.go @@ -13,6 +13,13 @@ import ( // share connections with the client being duplicated. It copies any headers already on the existing transport as // [codersdk.HeaderTransport] and add the headers in the argument. func DupClientCopyingHeaders(client *codersdk.Client, header http.Header) (*codersdk.Client, error) { + return DupClientConfiguringTransport(client, header, nil) +} + +// DupClientConfiguringTransport duplicates the Client like DupClientCopyingHeaders and, when configure is non-nil, +// calls it on the new transport before use. Callers that need to tune the connection pool go through this rather than +// reaching into the returned client, which would mean asserting a transport shape this function already has in hand. +func DupClientConfiguringTransport(client *codersdk.Client, header http.Header, configure func(*http.Transport)) (*codersdk.Client, error) { nc := codersdk.New(client.URL, codersdk.WithLogger(client.Logger())) nc.SessionTokenProvider = client.SessionTokenProvider newHeader, t, err := extractHeaderAndInnerTransport(client.HTTPClient.Transport) @@ -21,8 +28,13 @@ func DupClientCopyingHeaders(client *codersdk.Client, header http.Header) (*code } maps.Copy(newHeader, header) + transport := t.Clone() + if configure != nil { + configure(transport) + } + nc.HTTPClient.Transport = &codersdk.HeaderTransport{ - Transport: t.Clone(), + Transport: transport, Header: newHeader, } return nc, nil diff --git a/scaletest/notifications/config.go b/scaletest/notifications/config.go index 372199bf932..710c2ce06a5 100644 --- a/scaletest/notifications/config.go +++ b/scaletest/notifications/config.go @@ -2,30 +2,43 @@ package notifications import ( "net/http" + "net/url" "sync" "time" "github.com/google/uuid" "golang.org/x/xerrors" - "github.com/coder/coder/v2/scaletest/createusers" + "github.com/coder/coder/v2/codersdk" ) type Config struct { - // User is the configuration for the user to create. - User createusers.Config `json:"user"` - - // Roles are the roles to assign to the user. - Roles []string `json:"roles"` - - // NotificationTimeout is how long to wait for notifications after triggering. - NotificationTimeout time.Duration `json:"notification_timeout"` + // PreCreatedUser is the user the runner connects as. The caller must + // provide an already-authenticated user before the runner starts. + PreCreatedUser codersdk.User `json:"-"` + + // SessionToken authenticates PreCreatedUser for the websocket connection. + SessionToken string `json:"-"` + + // URL is the deployment address the runner dials. + URL *url.URL `json:"-"` + + // DialHTTPClient performs the websocket handshake request. It carries the + // caller's TLS and proxy configuration, without which the websocket library + // falls back to http.DefaultClient and ignores both. + // + // One client is shared by every runner. A websocket handshake returns 101 and + // hands the TCP connection to the caller, so it never re-enters the idle pool + // and every dial gets its own connection regardless of how many clients exist. + // It must not be a client whose transport caps MaxConnsPerHost, which would + // throttle the dials this test exists to make. + DialHTTPClient *http.Client `json:"-"` // DialTimeout is how long to wait for websocket connection. DialTimeout time.Duration `json:"dial_timeout"` - // ExpectedNotificationsIDs is the list of notification template IDs to expect. - ExpectedNotificationsIDs map[uuid.UUID]struct{} `json:"-"` + // ExpectedNotificationIDs is the set of notification template IDs to expect. + ExpectedNotificationIDs map[uuid.UUID]struct{} `json:"-"` Metrics *Metrics `json:"-"` @@ -35,7 +48,7 @@ type Config struct { // ReceivingWatchBarrier is the barrier for receiving users. Regular users wait on this to disconnect after receiving users complete. ReceivingWatchBarrier *sync.WaitGroup `json:"-"` - // SMTPApiUrl is the URL of the SMTP mock HTTP API + // SMTPApiURL is the URL of the SMTP mock HTTP API. SMTPApiURL string `json:"smtp_api_url"` // SMTPRequestTimeout is the timeout for SMTP requests. @@ -46,27 +59,30 @@ type Config struct { } func (c Config) Validate() error { - // The runner always needs an org; ensure we propagate it into the user config. - if c.User.OrganizationID == uuid.Nil { - return xerrors.New("user organization_id must be set") + if c.PreCreatedUser.ID == uuid.Nil { + return xerrors.New("pre_created_user must be set") + } + + if c.SessionToken == "" { + return xerrors.New("session_token must be set") } - if err := c.User.Validate(); err != nil { - return xerrors.Errorf("user config: %w", err) + if c.URL == nil { + return xerrors.New("url must be set") + } + + if c.DialHTTPClient == nil { + return xerrors.New("dial_http_client must be set") } if c.DialBarrier == nil { - return xerrors.New("dial barrier must be set") + return xerrors.New("dial_barrier must be set") } if c.ReceivingWatchBarrier == nil { return xerrors.New("receiving_watch_barrier must be set") } - if c.NotificationTimeout <= 0 { - return xerrors.New("notification_timeout must be greater than 0") - } - if c.SMTPApiURL != "" && c.SMTPRequestTimeout <= 0 { return xerrors.New("smtp_request_timeout must be set if smtp_api_url is set") } diff --git a/scaletest/notifications/config_test.go b/scaletest/notifications/config_test.go new file mode 100644 index 00000000000..6e03f2fcdd6 --- /dev/null +++ b/scaletest/notifications/config_test.go @@ -0,0 +1,138 @@ +package notifications_test + +import ( + "net/http" + "net/url" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + + "github.com/coder/coder/v2/codersdk" + "github.com/coder/coder/v2/scaletest/notifications" + "github.com/coder/coder/v2/testutil" +) + +// validConfig returns a Config that passes validation, so each case below can +// invalidate exactly one field and prove that field is checked. +func validConfig(t *testing.T) notifications.Config { + t.Helper() + + serverURL, err := url.Parse("http://coder.test") + require.NoError(t, err) + + return notifications.Config{ + PreCreatedUser: codersdk.User{ + ReducedUser: codersdk.ReducedUser{ + MinimalUser: codersdk.MinimalUser{ID: uuid.New()}, + }, + }, + SessionToken: "test-session-token", + URL: serverURL, + DialHTTPClient: &http.Client{}, + DialTimeout: testutil.WaitShort, + DialBarrier: new(sync.WaitGroup), + ReceivingWatchBarrier: new(sync.WaitGroup), + Metrics: notifications.NewMetrics(prometheus.NewRegistry()), + } +} + +func TestConfigValidate(t *testing.T) { + t.Parallel() + + t.Run("Valid", func(t *testing.T) { + t.Parallel() + + require.NoError(t, validConfig(t).Validate()) + }) + + t.Run("ValidWithSMTP", func(t *testing.T) { + t.Parallel() + + cfg := validConfig(t) + cfg.SMTPApiURL = "http://smtp.test" + cfg.SMTPRequestTimeout = testutil.WaitShort + cfg.SMTPHttpClient = &http.Client{} + require.NoError(t, cfg.Validate()) + }) + + for _, tc := range []struct { + name string + invalidate func(*notifications.Config) + errorContains string + }{ + { + name: "NoPreCreatedUser", + invalidate: func(c *notifications.Config) { c.PreCreatedUser = codersdk.User{} }, + errorContains: "pre_created_user must be set", + }, + { + name: "NoSessionToken", + invalidate: func(c *notifications.Config) { c.SessionToken = "" }, + errorContains: "session_token must be set", + }, + { + name: "NoURL", + invalidate: func(c *notifications.Config) { c.URL = nil }, + errorContains: "url must be set", + }, + { + name: "NoDialHTTPClient", + invalidate: func(c *notifications.Config) { c.DialHTTPClient = nil }, + errorContains: "dial_http_client must be set", + }, + { + name: "NoDialBarrier", + invalidate: func(c *notifications.Config) { c.DialBarrier = nil }, + errorContains: "dial_barrier must be set", + }, + { + name: "NoReceivingWatchBarrier", + invalidate: func(c *notifications.Config) { c.ReceivingWatchBarrier = nil }, + errorContains: "receiving_watch_barrier must be set", + }, + { + name: "NoDialTimeout", + invalidate: func(c *notifications.Config) { c.DialTimeout = 0 }, + errorContains: "dial_timeout must be greater than 0", + }, + { + name: "NegativeDialTimeout", + invalidate: func(c *notifications.Config) { c.DialTimeout = -time.Second }, + errorContains: "dial_timeout must be greater than 0", + }, + { + name: "NoMetrics", + invalidate: func(c *notifications.Config) { c.Metrics = nil }, + errorContains: "metrics must be set", + }, + { + // SMTP fields are only required once an SMTP URL is given. + name: "SMTPURLWithoutRequestTimeout", + invalidate: func(c *notifications.Config) { + c.SMTPApiURL = "http://smtp.test" + c.SMTPHttpClient = &http.Client{} + }, + errorContains: "smtp_request_timeout must be set if smtp_api_url is set", + }, + { + name: "SMTPURLWithoutHTTPClient", + invalidate: func(c *notifications.Config) { + c.SMTPApiURL = "http://smtp.test" + c.SMTPRequestTimeout = testutil.WaitShort + }, + errorContains: "smtp_http_client must be set if smtp_api_url is set", + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + cfg := validConfig(t) + tc.invalidate(&cfg) + require.ErrorContains(t, cfg.Validate(), tc.errorContains) + }) + } +} diff --git a/scaletest/notifications/run.go b/scaletest/notifications/run.go index bfc305d9744..95ac395f94f 100644 --- a/scaletest/notifications/run.go +++ b/scaletest/notifications/run.go @@ -19,7 +19,6 @@ import ( "cdr.dev/slog/v3/sloggers/sloghuman" "github.com/coder/coder/v2/coderd/tracing" "github.com/coder/coder/v2/codersdk" - "github.com/coder/coder/v2/scaletest/createusers" "github.com/coder/coder/v2/scaletest/harness" "github.com/coder/coder/v2/scaletest/loadtestutil" "github.com/coder/coder/v2/scaletest/smtpmock" @@ -28,10 +27,7 @@ import ( ) type Runner struct { - client *codersdk.Client - cfg Config - - createUserRunner *createusers.Runner + cfg Config // websocketReceiptTimes stores the receipt time for websocket notifications websocketReceiptTimes map[uuid.UUID]time.Time @@ -44,9 +40,11 @@ type Runner struct { clock quartz.Clock } -func NewRunner(client *codersdk.Client, cfg Config) *Runner { +// NewRunner returns a runner that dials the notification websocket as +// cfg.PreCreatedUser. It needs no API client: the caller owns the user lifecycle, +// so the handshake is the only request a runner makes. +func NewRunner(cfg Config) *Runner { return &Runner{ - client: client, cfg: cfg, websocketReceiptTimes: make(map[uuid.UUID]time.Time), smtpReceiptTimes: make(map[uuid.UUID]time.Time), @@ -61,11 +59,10 @@ func (r *Runner) WithClock(clock quartz.Clock) *Runner { var ( _ harness.Runnable = &Runner{} - _ harness.Cleanable = &Runner{} _ harness.Collectable = &Runner{} ) -func (r *Runner) Run(ctx context.Context, id string, logs io.Writer) error { +func (r *Runner) Run(ctx context.Context, _ string, logs io.Writer) error { ctx, span := tracing.StartSpan(ctx) defer span.End() @@ -78,49 +75,39 @@ func (r *Runner) Run(ctx context.Context, id string, logs io.Writer) error { reachedReceivingWatchBarrier := false defer func() { - if len(r.cfg.ExpectedNotificationsIDs) > 0 && !reachedReceivingWatchBarrier { + if len(r.cfg.ExpectedNotificationIDs) > 0 && !reachedReceivingWatchBarrier { r.cfg.ReceivingWatchBarrier.Done() } }() logs = loadtestutil.NewSyncWriter(logs) logger := slog.Make(sloghuman.Sink(logs)).Leveled(slog.LevelDebug) - r.client.SetLogger(logger) - r.client.SetLogBodies(true) - r.createUserRunner = createusers.NewRunner(r.client, r.cfg.User) - newUserAndToken, err := r.createUserRunner.RunReturningUser(ctx, id, logs) - if err != nil { - r.cfg.Metrics.AddError("create_user") - return xerrors.Errorf("create user: %w", err) + // Config.Validate owns this contract; these are defensive guards against a + // caller that bypasses validation. A caller-side programming error is not a + // load-test failure, so neither is recorded as an error metric. + if r.cfg.PreCreatedUser.ID == uuid.Nil { + return xerrors.New("pre-created user required but not provided") } - newUser := newUserAndToken.User - newUserClient := codersdk.New(r.client.URL, - codersdk.WithSessionToken(newUserAndToken.SessionToken), + if r.cfg.SessionToken == "" { + return xerrors.New("session token required but not provided") + } + user := r.cfg.PreCreatedUser + userClient := codersdk.New(r.cfg.URL, + codersdk.WithSessionToken(r.cfg.SessionToken), codersdk.WithLogger(logger), codersdk.WithLogBodies()) + // Dial with the caller's HTTP client so the handshake uses its TLS and proxy + // configuration. + userClient.HTTPClient = r.cfg.DialHTTPClient - logger.Info(ctx, "runner user created", slog.F("username", newUser.Username), slog.F("user_id", newUser.ID.String())) - - if len(r.cfg.Roles) > 0 { - logger.Info(ctx, "assigning roles to user", slog.F("roles", r.cfg.Roles)) - - _, err := r.client.UpdateUserRoles(ctx, newUser.ID.String(), codersdk.UpdateRoles{ - Roles: r.cfg.Roles, - }) - if err != nil { - r.cfg.Metrics.AddError("assign_roles") - return xerrors.Errorf("assign roles: %w", err) - } - } - - logger.Info(ctx, "notification runner is ready") + logger.Info(ctx, "notification runner is ready", slog.F("username", user.Username), slog.F("user_id", user.ID.String())) dialCtx, cancel := context.WithTimeout(ctx, r.cfg.DialTimeout) defer cancel() logger.Info(ctx, "connecting to notification websocket") - conn, err := r.dialNotificationWebsocket(dialCtx, newUserClient, logger) + conn, err := r.dialNotificationWebsocket(dialCtx, userClient, logger) if err != nil { return xerrors.Errorf("dial notification websocket: %w", err) } @@ -131,7 +118,7 @@ func (r *Runner) Run(ctx context.Context, id string, logs io.Writer) error { r.cfg.DialBarrier.Done() r.cfg.DialBarrier.Wait() - if len(r.cfg.ExpectedNotificationsIDs) == 0 { + if len(r.cfg.ExpectedNotificationIDs) == 0 { logger.Info(ctx, "maintaining websocket connection, waiting for receiving users to complete") // Wait for receiving users to complete @@ -150,21 +137,25 @@ func (r *Runner) Run(ctx context.Context, id string, logs io.Writer) error { return nil } - logger.Info(ctx, "waiting for notifications", slog.F("timeout", r.cfg.NotificationTimeout)) - - watchCtx, cancel := context.WithTimeout(ctx, r.cfg.NotificationTimeout) - defer cancel() + // The watch runs until the caller's context expires. That context carries the + // overall test budget, so there is no separate per-runner notification + // deadline that could expire before or after it. + if deadline, ok := ctx.Deadline(); ok { + logger.Info(ctx, "waiting for notifications", slog.F("deadline", deadline)) + } else { + logger.Info(ctx, "waiting for notifications") + } - eg, egCtx := errgroup.WithContext(watchCtx) + eg, egCtx := errgroup.WithContext(ctx) eg.Go(func() error { - return r.watchNotifications(egCtx, conn, newUser, logger, r.cfg.ExpectedNotificationsIDs) + return r.watchNotifications(egCtx, conn, user, logger, r.cfg.ExpectedNotificationIDs) }) if r.cfg.SMTPApiURL != "" { logger.Info(ctx, "running SMTP notification watcher") eg.Go(func() error { - return r.watchNotificationsSMTP(egCtx, newUser, logger, r.cfg.ExpectedNotificationsIDs) + return r.watchNotificationsSMTP(egCtx, user, logger, r.cfg.ExpectedNotificationIDs) }) } @@ -178,17 +169,6 @@ func (r *Runner) Run(ctx context.Context, id string, logs io.Writer) error { return nil } -func (r *Runner) Cleanup(ctx context.Context, id string, logs io.Writer) error { - if r.createUserRunner != nil { - _, _ = fmt.Fprintln(logs, "Cleaning up user...") - if err := r.createUserRunner.Cleanup(ctx, id, logs); err != nil { - return xerrors.Errorf("cleanup user: %w", err) - } - } - - return nil -} - const ( WebsocketNotificationReceiptTimeMetric = "notification_websocket_receipt_time" SMTPNotificationReceiptTimeMetric = "notification_smtp_receipt_time" @@ -209,31 +189,16 @@ func (r *Runner) GetMetrics() map[string]any { } } +// dialNotificationWebsocket connects to the inbox watch endpoint. Client.Dial +// parses the URL, propagates the client's HTTP client so custom TLS applies, and +// sets the session token header. func (r *Runner) dialNotificationWebsocket(ctx context.Context, client *codersdk.Client, logger slog.Logger) (*websocket.Conn, error) { - u, err := client.URL.Parse("/api/v2/notifications/inbox/watch") + conn, err := client.Dial(ctx, "/api/v2/notifications/inbox/watch", nil) if err != nil { - logger.Error(ctx, "parse notification URL", slog.Error(err)) - r.cfg.Metrics.AddError("parse_url") - return nil, xerrors.Errorf("parse notification URL: %w", err) - } - - conn, resp, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{ - HTTPHeader: http.Header{ - "Coder-Session-Token": []string{client.SessionToken()}, - }, - }) - if err != nil { - if resp != nil { - defer resp.Body.Close() - if resp.StatusCode != http.StatusSwitchingProtocols { - err = codersdk.ReadBodyAsError(resp) - } - } logger.Error(ctx, "dial notification websocket", slog.Error(err)) r.cfg.Metrics.AddError("dial") return nil, xerrors.Errorf("dial notification websocket: %w", err) } - return conn, nil } @@ -347,6 +312,10 @@ func (r *Runner) watchNotificationsSMTP(ctx context.Context, user codersdk.User, if _, exists := expectedNotifications[notificationID]; exists { if _, received := receivedNotifications[notificationID]; !received { + // The SMTP mock stamps this date on its own host, while the trigger time + // comes from the CLI host, so clock skew between the two lands in the + // reported SMTP latency and can even make it negative. The websocket + // measurement stays on one clock and does not have this problem. receiptTime := summary.Date if receiptTime.IsZero() { receiptTime = time.Now() diff --git a/scaletest/notifications/run_test.go b/scaletest/notifications/run_test.go index a9ef6f4b296..9fe3cf980c8 100644 --- a/scaletest/notifications/run_test.go +++ b/scaletest/notifications/run_test.go @@ -22,8 +22,6 @@ import ( notificationsLib "github.com/coder/coder/v2/coderd/notifications" "github.com/coder/coder/v2/coderd/notifications/dispatch" "github.com/coder/coder/v2/coderd/notifications/types" - "github.com/coder/coder/v2/codersdk" - "github.com/coder/coder/v2/scaletest/createusers" "github.com/coder/coder/v2/scaletest/notifications" "github.com/coder/coder/v2/scaletest/smtpmock" "github.com/coder/coder/v2/testutil" @@ -47,78 +45,72 @@ func TestRun(t *testing.T) { const numReceivingUsers = 2 const numRegularUsers = 2 + const totalUsers = numReceivingUsers + numRegularUsers + metrics := notifications.NewMetrics(prometheus.NewRegistry()) + + // The generator triggers a single template-deleted notification, so that is + // the notification the receiving runners expect. + expectedNotificationIDs := map[uuid.UUID]struct{}{ + notificationsLib.TemplateTemplateDeleted: {}, + } + dialBarrier := new(sync.WaitGroup) receivingWatchBarrier := new(sync.WaitGroup) - dialBarrier.Add(numReceivingUsers + numRegularUsers) + dialBarrier.Add(totalUsers) receivingWatchBarrier.Add(numReceivingUsers) - metrics := notifications.NewMetrics(prometheus.NewRegistry()) eg, runCtx := errgroup.WithContext(ctx) - expectedNotificationsIDs := map[uuid.UUID]struct{}{ - notificationsLib.TemplateUserAccountCreated: {}, - notificationsLib.TemplateUserAccountDeleted: {}, - } - - // Start receiving runners who will receive notifications receivingRunners := make([]*notifications.Runner, 0, numReceivingUsers) + receivingUsernames := make([]string, 0, numReceivingUsers) for i := range numReceivingUsers { + userClient, user := coderdtest.CreateAnotherUser(t, client, firstUser.OrganizationID) + receivingUsernames = append(receivingUsernames, user.Username) runnerCfg := notifications.Config{ - User: createusers.Config{ - OrganizationID: firstUser.OrganizationID, - Username: "receiving-user-" + strconv.Itoa(i), - }, - Roles: []string{codersdk.RoleOwner}, - NotificationTimeout: testutil.WaitLong, - DialTimeout: testutil.WaitLong, - Metrics: metrics, - DialBarrier: dialBarrier, - ReceivingWatchBarrier: receivingWatchBarrier, - ExpectedNotificationsIDs: expectedNotificationsIDs, + PreCreatedUser: user, + SessionToken: userClient.SessionToken(), + URL: client.URL, + DialHTTPClient: client.HTTPClient, + DialTimeout: testutil.WaitLong, + Metrics: metrics, + DialBarrier: dialBarrier, + ReceivingWatchBarrier: receivingWatchBarrier, + ExpectedNotificationIDs: expectedNotificationIDs, } - err := runnerCfg.Validate() - require.NoError(t, err) + require.NoError(t, runnerCfg.Validate()) - runner := notifications.NewRunner(client, runnerCfg) + runner := notifications.NewRunner(runnerCfg) receivingRunners = append(receivingRunners, runner) eg.Go(func() error { return runner.Run(runCtx, "receiving-"+strconv.Itoa(i), io.Discard) }) } - // Start regular user runners who will maintain websocket connections - regularRunners := make([]*notifications.Runner, 0, numRegularUsers) for i := range numRegularUsers { + userClient, user := coderdtest.CreateAnotherUser(t, client, firstUser.OrganizationID) runnerCfg := notifications.Config{ - User: createusers.Config{ - OrganizationID: firstUser.OrganizationID, - }, - Roles: []string{}, - NotificationTimeout: testutil.WaitLong, + PreCreatedUser: user, + SessionToken: userClient.SessionToken(), + URL: client.URL, + DialHTTPClient: client.HTTPClient, DialTimeout: testutil.WaitLong, Metrics: metrics, DialBarrier: dialBarrier, ReceivingWatchBarrier: receivingWatchBarrier, } - err := runnerCfg.Validate() - require.NoError(t, err) + require.NoError(t, runnerCfg.Validate()) - runner := notifications.NewRunner(client, runnerCfg) - regularRunners = append(regularRunners, runner) + runner := notifications.NewRunner(runnerCfg) eg.Go(func() error { return runner.Run(runCtx, "regular-"+strconv.Itoa(i), io.Discard) }) } - // Trigger notifications by creating and deleting a user eg.Go(func() error { - // Wait for all runners to connect dialBarrier.Wait() - for i := 0; i < numReceivingUsers; i++ { - err := sendInboxNotification(runCtx, t, db, inboxHandler, "receiving-user-"+strconv.Itoa(i), notificationsLib.TemplateUserAccountCreated) - require.NoError(t, err) - err = sendInboxNotification(runCtx, t, db, inboxHandler, "receiving-user-"+strconv.Itoa(i), notificationsLib.TemplateUserAccountDeleted) + for _, username := range receivingUsernames { + err := sendInboxNotification(runCtx, t, db, inboxHandler, username, notificationsLib.TemplateTemplateDeleted) require.NoError(t, err) } @@ -128,31 +120,11 @@ func TestRun(t *testing.T) { err := eg.Wait() require.NoError(t, err, "runner execution should complete successfully") - cleanupEg, cleanupCtx := errgroup.WithContext(ctx) - for i, runner := range receivingRunners { - cleanupEg.Go(func() error { - return runner.Cleanup(cleanupCtx, "receiving-"+strconv.Itoa(i), io.Discard) - }) - } - for i, runner := range regularRunners { - cleanupEg.Go(func() error { - return runner.Cleanup(cleanupCtx, "regular-"+strconv.Itoa(i), io.Discard) - }) - } - err = cleanupEg.Wait() - require.NoError(t, err) - - users, err := client.Users(ctx, codersdk.UsersRequest{}) - require.NoError(t, err) - require.Len(t, users.Users, 1) - require.Equal(t, firstUser.UserID, users.Users[0].ID) - for _, runner := range receivingRunners { metrics := runner.GetMetrics() websocketReceiptTimes := metrics[notifications.WebsocketNotificationReceiptTimeMetric].(map[uuid.UUID]time.Time) - require.Contains(t, websocketReceiptTimes, notificationsLib.TemplateUserAccountCreated) - require.Contains(t, websocketReceiptTimes, notificationsLib.TemplateUserAccountDeleted) + require.Contains(t, websocketReceiptTimes, notificationsLib.TemplateTemplateDeleted) } } @@ -175,14 +147,9 @@ func TestRunWithSMTP(t *testing.T) { smtpAPIMux.HandleFunc("/messages", func(w http.ResponseWriter, r *http.Request) { summaries := []smtpmock.EmailSummary{ { - Subject: "TemplateUserAccountCreated", - Date: time.Now(), - NotificationTemplateID: notificationsLib.TemplateUserAccountCreated, - }, - { - Subject: "TemplateUserAccountDeleted", + Subject: "TemplateTemplateDeleted", Date: time.Now(), - NotificationTemplateID: notificationsLib.TemplateUserAccountDeleted, + NotificationTemplateID: notificationsLib.TemplateTemplateDeleted, }, } @@ -195,17 +162,11 @@ func TestRunWithSMTP(t *testing.T) { const numReceivingUsers = 2 const numRegularUsers = 2 - dialBarrier := new(sync.WaitGroup) - receivingWatchBarrier := new(sync.WaitGroup) - dialBarrier.Add(numReceivingUsers + numRegularUsers) - receivingWatchBarrier.Add(numReceivingUsers) + const totalUsers = numReceivingUsers + numRegularUsers metrics := notifications.NewMetrics(prometheus.NewRegistry()) - eg, runCtx := errgroup.WithContext(ctx) - - expectedNotificationsIDs := map[uuid.UUID]struct{}{ - notificationsLib.TemplateUserAccountCreated: {}, - notificationsLib.TemplateUserAccountDeleted: {}, + expectedNotificationIDs := map[uuid.UUID]struct{}{ + notificationsLib.TemplateTemplateDeleted: {}, } mClock := quartz.NewMock(t) @@ -214,72 +175,70 @@ func TestRunWithSMTP(t *testing.T) { httpClient := &http.Client{} - // Start receiving runners who will receive notifications + dialBarrier := new(sync.WaitGroup) + receivingWatchBarrier := new(sync.WaitGroup) + dialBarrier.Add(totalUsers) + receivingWatchBarrier.Add(numReceivingUsers) + + eg, runCtx := errgroup.WithContext(ctx) + receivingRunners := make([]*notifications.Runner, 0, numReceivingUsers) + receivingUsernames := make([]string, 0, numReceivingUsers) for i := range numReceivingUsers { + userClient, user := coderdtest.CreateAnotherUser(t, client, firstUser.OrganizationID) + receivingUsernames = append(receivingUsernames, user.Username) runnerCfg := notifications.Config{ - User: createusers.Config{ - OrganizationID: firstUser.OrganizationID, - Username: "receiving-user-" + strconv.Itoa(i), - }, - Roles: []string{codersdk.RoleOwner}, - NotificationTimeout: testutil.WaitLong, - DialTimeout: testutil.WaitLong, - Metrics: metrics, - DialBarrier: dialBarrier, - ReceivingWatchBarrier: receivingWatchBarrier, - ExpectedNotificationsIDs: expectedNotificationsIDs, - SMTPApiURL: smtpAPIServer.URL, - SMTPRequestTimeout: testutil.WaitLong, - SMTPHttpClient: httpClient, + PreCreatedUser: user, + SessionToken: userClient.SessionToken(), + URL: client.URL, + DialHTTPClient: client.HTTPClient, + DialTimeout: testutil.WaitLong, + Metrics: metrics, + DialBarrier: dialBarrier, + ReceivingWatchBarrier: receivingWatchBarrier, + ExpectedNotificationIDs: expectedNotificationIDs, + SMTPApiURL: smtpAPIServer.URL, + SMTPRequestTimeout: testutil.WaitLong, + SMTPHttpClient: httpClient, } - err := runnerCfg.Validate() - require.NoError(t, err) + require.NoError(t, runnerCfg.Validate()) - runner := notifications.NewRunner(client, runnerCfg).WithClock(mClock) + runner := notifications.NewRunner(runnerCfg).WithClock(mClock) receivingRunners = append(receivingRunners, runner) eg.Go(func() error { return runner.Run(runCtx, "receiving-"+strconv.Itoa(i), io.Discard) }) } - // Start regular user runners who will maintain websocket connections - regularRunners := make([]*notifications.Runner, 0, numRegularUsers) for i := range numRegularUsers { + userClient, user := coderdtest.CreateAnotherUser(t, client, firstUser.OrganizationID) runnerCfg := notifications.Config{ - User: createusers.Config{ - OrganizationID: firstUser.OrganizationID, - }, - Roles: []string{}, - NotificationTimeout: testutil.WaitLong, + PreCreatedUser: user, + SessionToken: userClient.SessionToken(), + URL: client.URL, + DialHTTPClient: client.HTTPClient, DialTimeout: testutil.WaitLong, Metrics: metrics, DialBarrier: dialBarrier, ReceivingWatchBarrier: receivingWatchBarrier, } - err := runnerCfg.Validate() - require.NoError(t, err) + require.NoError(t, runnerCfg.Validate()) - runner := notifications.NewRunner(client, runnerCfg) - regularRunners = append(regularRunners, runner) + runner := notifications.NewRunner(runnerCfg) eg.Go(func() error { return runner.Run(runCtx, "regular-"+strconv.Itoa(i), io.Discard) }) } - // Trigger notifications by creating and deleting a user eg.Go(func() error { - // Wait for all runners to connect dialBarrier.Wait() - for i := 0; i < numReceivingUsers; i++ { + for range receivingUsernames { smtpTrap.MustWait(runCtx).MustRelease(runCtx) } - for i := 0; i < numReceivingUsers; i++ { - err := sendInboxNotification(runCtx, t, db, inboxHandler, "receiving-user-"+strconv.Itoa(i), notificationsLib.TemplateUserAccountCreated) - require.NoError(t, err) - err = sendInboxNotification(runCtx, t, db, inboxHandler, "receiving-user-"+strconv.Itoa(i), notificationsLib.TemplateUserAccountDeleted) + for _, username := range receivingUsernames { + err := sendInboxNotification(runCtx, t, db, inboxHandler, username, notificationsLib.TemplateTemplateDeleted) require.NoError(t, err) } @@ -292,35 +251,13 @@ func TestRunWithSMTP(t *testing.T) { err := eg.Wait() require.NoError(t, err, "runner execution with SMTP should complete successfully") - cleanupEg, cleanupCtx := errgroup.WithContext(ctx) - for i, runner := range receivingRunners { - cleanupEg.Go(func() error { - return runner.Cleanup(cleanupCtx, "receiving-"+strconv.Itoa(i), io.Discard) - }) - } - for i, runner := range regularRunners { - cleanupEg.Go(func() error { - return runner.Cleanup(cleanupCtx, "regular-"+strconv.Itoa(i), io.Discard) - }) - } - err = cleanupEg.Wait() - require.NoError(t, err) - - users, err := client.Users(ctx, codersdk.UsersRequest{}) - require.NoError(t, err) - require.Len(t, users.Users, 1) - require.Equal(t, firstUser.UserID, users.Users[0].ID) - - // Verify that notifications were received via both websocket and SMTP for _, runner := range receivingRunners { metrics := runner.GetMetrics() websocketReceiptTimes := metrics[notifications.WebsocketNotificationReceiptTimeMetric].(map[uuid.UUID]time.Time) smtpReceiptTimes := metrics[notifications.SMTPNotificationReceiptTimeMetric].(map[uuid.UUID]time.Time) - require.Contains(t, websocketReceiptTimes, notificationsLib.TemplateUserAccountCreated) - require.Contains(t, websocketReceiptTimes, notificationsLib.TemplateUserAccountDeleted) - require.Contains(t, smtpReceiptTimes, notificationsLib.TemplateUserAccountCreated) - require.Contains(t, smtpReceiptTimes, notificationsLib.TemplateUserAccountDeleted) + require.Contains(t, websocketReceiptTimes, notificationsLib.TemplateTemplateDeleted) + require.Contains(t, smtpReceiptTimes, notificationsLib.TemplateTemplateDeleted) } } @@ -345,3 +282,61 @@ func sendInboxNotification(ctx context.Context, t *testing.T, db database.Store, return nil } + +// TestRunNotificationNeverArrives covers the runner's timeout path: an expected +// notification that never arrives must fail the run rather than return nil. This +// PR re-timed that path, replacing the per-runner notification timeout with the +// caller's context, so the failure now depends on the context alone. +func TestRunNotificationNeverArrives(t *testing.T) { + t.Parallel() + + ctx := testutil.Context(t, testutil.WaitLong) + client := coderdtest.New(t, nil) + firstUser := coderdtest.CreateFirstUser(t, client) + metrics := notifications.NewMetrics(prometheus.NewRegistry()) + + userClient, user := coderdtest.CreateAnotherUser(t, client, firstUser.OrganizationID) + + dialBarrier := new(sync.WaitGroup) + receivingWatchBarrier := new(sync.WaitGroup) + dialBarrier.Add(1) + receivingWatchBarrier.Add(1) + + runnerCfg := notifications.Config{ + PreCreatedUser: user, + SessionToken: userClient.SessionToken(), + URL: client.URL, + DialHTTPClient: client.HTTPClient, + DialTimeout: testutil.WaitLong, + Metrics: metrics, + DialBarrier: dialBarrier, + ReceivingWatchBarrier: receivingWatchBarrier, + // Expect a notification that nothing ever sends. + ExpectedNotificationIDs: map[uuid.UUID]struct{}{ + notificationsLib.TemplateTemplateDeleted: {}, + }, + } + require.NoError(t, runnerCfg.Validate()) + + // Cancel once the runner has connected, so the dial is never the thing that + // fails and the watch is the only step left. The runner releases the dial + // barrier immediately after connecting. + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + go func() { + dialBarrier.Wait() + cancel() + }() + + runner := notifications.NewRunner(runnerCfg) + err := runner.Run(runCtx, "receiving-0", io.Discard) + // Assert the specific path: Run wraps every watch failure with the same prefix, + // so matching only that would also accept a read or SMTP failure. + require.ErrorIs(t, err, context.Canceled, + "a notification that never arrives must fail the run on the canceled context") + + // No receipt time is recorded, so the run contributes no latency rather than a + // bogus one. + receiptTimes := runner.GetMetrics()[notifications.WebsocketNotificationReceiptTimeMetric].(map[uuid.UUID]time.Time) + require.Empty(t, receiptTimes) +}