Method watchWorkspaceUpdates

scaletest/taskstatus/run.go:158–209

(ctx context.Context)

}

func (r *Runner) watchWorkspaceUpdates(ctx context.Context) error {
	shouldMarkConnectedDone := true
	defer func() {
		if shouldMarkConnectedDone {
			r.cfg.ConnectedWaitGroup.Done()
		}
	}()
	updates, err := r.client.watchWorkspace(ctx, r.workspaceID)
	if err != nil {
		return xerrors.Errorf("watch workspace: %w", err)
	}
	shouldMarkConnectedDone = false
	r.cfg.ConnectedWaitGroup.Done()
	defer func() {
		r.mu.Lock()
		defer r.mu.Unlock()
		r.cfg.Metrics.MissingStatusUpdatesTotal.
			WithLabelValues(r.cfg.MetricLabelValues...).
			Add(float64(len(r.reportTimes)))
	}()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case workspace := <-updates:
			if workspace.LatestAppStatus == nil {
				continue
			}
			msgNo, ok := parseStatusMessage(workspace.LatestAppStatus.Message)
			if !ok {
				continue
			}

			r.mu.Lock()
			reportTime, ok := r.reportTimes[msgNo]
			delete(r.reportTimes, msgNo)
			allDone := r.doneReporting && len(r.reportTimes) == 0
			r.mu.Unlock()

			if !ok {
				return xerrors.Errorf("report time not found for message %d", msgNo)
			}
			latency := r.clock.Since(reportTime, "watchWorkspaceUpdates")
			r.cfg.Metrics.TaskStatusToWorkspaceUpdateLatencySeconds.
				WithLabelValues(r.cfg.MetricLabelValues...).
				Observe(latency.Seconds())
			if allDone {
				return nil
			}
		}
	}
}

func (r *Runner) reportTaskStatus(ctx context.Context) error {
	defer func() {