watchWorkspaceUpdates processes workspace updates and returns error or nil once all expected workspaces and agents are seen.
(ctx context.Context, clients *tailnet.ControlProtocolClients, user codersdk.User, logger slog.Logger)
| 200 | // watchWorkspaceUpdates processes workspace updates and returns error or nil |
| 201 | // once all expected workspaces and agents are seen. |
| 202 | func (r *Runner) watchWorkspaceUpdates(ctx context.Context, clients *tailnet.ControlProtocolClients, user codersdk.User, logger slog.Logger) error { |
| 203 | expectedWorkspaces := r.cfg.WorkspaceCount |
| 204 | // workspace name to time the update was seen |
| 205 | seenWorkspaces := make(map[string]time.Time) |
| 206 | |
| 207 | logger.Info(ctx, fmt.Sprintf("waiting for %d workspaces and their agents", expectedWorkspaces)) |
| 208 | for { |
| 209 | select { |
| 210 | case <-ctx.Done(): |
| 211 | logger.Error(ctx, "context canceled while waiting for workspace updates", slog.Error(ctx.Err())) |
| 212 | r.cfg.Metrics.AddError(user.Username, r.cfg.WorkspaceCount, "context_done") |
| 213 | return ctx.Err() |
| 214 | default: |
| 215 | } |
| 216 | |
| 217 | update, err := clients.WorkspaceUpdates.Recv() |
| 218 | if err != nil { |
| 219 | logger.Error(ctx, "workspace updates stream error", slog.Error(err)) |
| 220 | r.cfg.Metrics.AddError(user.Username, r.cfg.WorkspaceCount, "recv") |
| 221 | return xerrors.Errorf("receive workspace update: %w", err) |
| 222 | } |
| 223 | recvTime := time.Now() |
| 224 | |
| 225 | for _, ws := range update.UpsertedWorkspaces { |
| 226 | seenWorkspaces[ws.Name] = recvTime |
| 227 | } |
| 228 | |
| 229 | if len(seenWorkspaces) == int(expectedWorkspaces) { |
| 230 | for wsName, seenTime := range seenWorkspaces { |
| 231 | // We only receive workspace updates for those that we built. |
| 232 | // If we received a workspace update for a workspace we didn't build, |
| 233 | // we're risking racing with the code that writes workspace |
| 234 | // build start times to this map. |
| 235 | ws, ok := r.workspaces[wsName] |
| 236 | if !ok { |
| 237 | logger.Error(ctx, "received update for unexpected workspace", slog.F("workspace", wsName), slog.F("seen_workspaces", seenWorkspaces)) |
| 238 | r.cfg.Metrics.AddError(user.Username, r.cfg.WorkspaceCount, "unexpected_workspace") |
| 239 | return xerrors.Errorf("received update for unexpected workspace %q", wsName) |
| 240 | } |
| 241 | ws.updateLatency = seenTime.Sub(ws.buildStartTime) |
| 242 | r.cfg.Metrics.RecordCompletion(ws.updateLatency, user.Username, r.cfg.WorkspaceCount, wsName) |
| 243 | } |
| 244 | logger.Info(ctx, fmt.Sprintf("updates received for all %d workspaces and agents", expectedWorkspaces)) |
| 245 | return nil |
| 246 | } |
| 247 | } |
| 248 | } |
| 249 | |
| 250 | const ( |
| 251 | WorkspaceUpdatesLatencyMetric = "workspace_updates_latency_seconds" |