| 66 | } |
| 67 | |
| 68 | func (r *Runner) Run(ctx context.Context, name string, logs io.Writer) error { |
| 69 | shouldMarkConnectedDone := true |
| 70 | defer func() { |
| 71 | if shouldMarkConnectedDone { |
| 72 | r.cfg.ConnectedWaitGroup.Done() |
| 73 | } |
| 74 | }() |
| 75 | |
| 76 | // ensure these labels are initialized, so we see the time series right away in prometheus. |
| 77 | r.cfg.Metrics.MissingStatusUpdatesTotal.WithLabelValues(r.cfg.MetricLabelValues...).Add(0) |
| 78 | r.cfg.Metrics.ReportTaskStatusErrorsTotal.WithLabelValues(r.cfg.MetricLabelValues...).Add(0) |
| 79 | |
| 80 | logs = loadtestutil.NewSyncWriter(logs) |
| 81 | r.logger = slog.Make(sloghuman.Sink(logs)).Leveled(slog.LevelDebug).Named(name) |
| 82 | r.client.initialize(r.logger) |
| 83 | |
| 84 | // Create the external workspace |
| 85 | r.logger.Info(ctx, "creating external workspace", |
| 86 | slog.F("template_id", r.cfg.TemplateID), |
| 87 | slog.F("workspace_name", r.cfg.WorkspaceName)) |
| 88 | |
| 89 | result, err := r.createExternalWorkspace(ctx, codersdk.CreateWorkspaceRequest{ |
| 90 | TemplateID: r.cfg.TemplateID, |
| 91 | Name: r.cfg.WorkspaceName, |
| 92 | }) |
| 93 | if err != nil { |
| 94 | r.cfg.Metrics.ReportTaskStatusErrorsTotal.WithLabelValues(r.cfg.MetricLabelValues...).Inc() |
| 95 | return xerrors.Errorf("create external workspace: %w", err) |
| 96 | } |
| 97 | |
| 98 | // Set the workspace ID |
| 99 | r.workspaceID = result.workspaceID |
| 100 | r.logger.Info(ctx, "created external workspace", slog.F("workspace_id", r.workspaceID)) |
| 101 | |
| 102 | // Establish the dRPC connection using the agent token. |
| 103 | if err := r.updater.initialize(ctx, r.logger, result.agentToken); err != nil { |
| 104 | r.cfg.Metrics.ReportTaskStatusErrorsTotal.WithLabelValues(r.cfg.MetricLabelValues...).Inc() |
| 105 | return xerrors.Errorf("initialize app status updater: %w", err) |
| 106 | } |
| 107 | defer func() { |
| 108 | if err := r.updater.close(); err != nil { |
| 109 | r.logger.Error(ctx, "failed to close app status updater", slog.Error(err)) |
| 110 | } |
| 111 | }() |
| 112 | r.logger.Info(ctx, "initialized app status updater with agent token") |
| 113 | |
| 114 | workspaceUpdatesCtx, cancelWorkspaceUpdates := context.WithCancel(ctx) |
| 115 | defer cancelWorkspaceUpdates() |
| 116 | workspaceUpdatesResult := make(chan error, 1) |
| 117 | shouldMarkConnectedDone = false // we are passing this responsibility to the watchWorkspaceUpdates goroutine |
| 118 | go func() { |
| 119 | workspaceUpdatesResult <- r.watchWorkspaceUpdates(workspaceUpdatesCtx) |
| 120 | }() |
| 121 | |
| 122 | err = r.reportTaskStatus(ctx) |
| 123 | if err != nil { |
| 124 | return xerrors.Errorf("report task status: %w", err) |
| 125 | } |