Method handleRequestQueue

pkg/workerpool/worker_pool.go:245–286

()

}

func (w *WorkerPool) handleRequestQueue() {
	w.Lock()
	defer w.Unlock()
	klog.V(4).InfoS("handle request queue")
	for _, item := range w.requestQueue.All() {
		req := item.(Request)
		if _, err := w.matchWorkerFor(&req); err != nil {
			if err != ErrNotWorker {
				klog.ErrorS(err, "failed to create worker for cloudshell", req.Cloudshell)
			}

			pods, err := w.podLister.Pods(req.Namespace).List(labels.Set{constants.WorkerRequestLabelKey: req.Cloudshell}.AsSelector())
			if err != nil {
				klog.ErrorS(err, "failed to list worker for cloudshell", req.Cloudshell)
			}

			// TODO: pod is failed?

			if len(pods) == 0 {
				klog.InfoS("start to create worker", "cloudshell", req.Cloudshell)
				if err := w.createWorker(&req); err != nil {
					klog.ErrorS(err, "failed to create worker for cloudshell", req.Cloudshell)
				}
			}
		} else {
			cloudshell := &cloudshellv1alpha1.CloudShell{}
			if err = w.Get(context.TODO(), client.ObjectKey{Namespace: req.Namespace, Name: req.Cloudshell}, cloudshell); err != nil {
				if apierrors.IsNotFound(err) {
					w.requestQueue.Remove(req)
				}

				klog.ErrorS(err, "failed to get cloudshell", req.Cloudshell)
				continue
			}

			key, _ := cache.MetaNamespaceKeyFunc(cloudshell)
			req.CloudShellQueue.Add(key)

			w.requestQueue.Remove(req)
		}
	}
}

func (w *WorkerPool) tryScaleInWorkerQueue(stop <-chan struct{}) {
	t := time.NewTicker(w.scaleInQueueDuration)