} else {
			// If oldpod != ready && newpod != ready, but the ambient annotation was added,
			// then assume this event was generated by the CNI plugin labeling the pod on startup,
			// and skip the event.
			//
			// This isn't perfect (someone could manually annotate an unready pod,
			// then install Istio, then the pod goes ready, and we'd miss capture) - but that
			// seems vanishingly unlikely

		},
		expectedQueueLen: 1,
	}, {
		name: "lopsided but 1 unready, queue required because unready node means 0 CPU in one zone",
		nodes: []nodeInfo{
			{zoneLabel: &zoneA, ready: &readyFalse, cpu: &cpu100},
			{zoneLabel: &zoneB, ready: &readyTrue, cpu: &cpu1000},
			{zoneLabel: &zoneC, ready: &readyTrue, cpu: &cpu2000},
		},
		topologyCacheEnabled: true,

	// Before accessing the cacher's cache, wait for the ready to be ok.
	// This is necessary to prevent users from accessing structures that are
	// uninitialized or are being repopulated right now.
	// ready needs to be set to false when the cacher is paused or stopped.
	// ready needs to be set to true when the cacher is ready to use after
	// initialization.
	ready *ready

	// Underlying storage.Interface.

			handleProbeSync(kl, update, handler, "liveness", "unhealthy")
		}
	case update := <-kl.readinessManager.Updates():
		ready := update.Result == proberesults.Success
		kl.statusManager.SetContainerReadiness(update.PodUID, update.ContainerID, ready)

		status := ""
		if ready {
			status = "ready"
		}
		handleProbeSync(kl, update, handler, "readiness", status)
	case update := <-kl.startupManager.Updates():

	if err != nil {
		return nil, fmt.Errorf("error starting cni server: %w", err)
	}
	s.cniServerStopFunc = cniServer.Stop

	return s, nil
}

func (s *Server) Ready() {
	s.isReady.Store(true)
}

func (s *Server) NotReady() {
	s.isReady.Store(false)
}

// buildKubeClient creates the kube client
func buildKubeClient(kubeConfig string) (kube.Client, error) {
	// Used by validation

     * the thread pool is full of work items that are waiting on other queued work items.  The queued work items cannot execute
     * because the thread pool is already full with their parent work items.  We use expand() to allow the thread pool to temporarily
     * expand when work items have to wait on other work.  The thread pool will shrink below max workers again once the queue is
     * drained.

	// value of wait because any previous waiters are already in the list
	// or will notice that they have already been notified when trying to
	// add themselves to the list.
	atomic.Store(&l.notify, l.wait.Load())
	unlock(&l.lock)

	// Go through the local list and ready all waiters.
	for s != nil {
		next := s.next
		s.next = nil
		readyWithTime(s, 4)
		s = next

		// Add new endpoints only if they are ever ready once to shards
		// so that full push does not send them from shards.
		for _, oie := range oldIstioEndpoints {
			omap[oie.Address] = oie
		}
		for _, nie := range istioEndpoints {
			nmap[nie.Address] = nie
		}
		for _, nie := range istioEndpoints {
			if oie, exists := omap[nie.Address]; exists {
				// If endpoint exists already, we should push if it's changed.

// dot generates a web page containing an svg diagram.
func (ui *webInterface) dot(w http.ResponseWriter, req *http.Request) {
	rpt, errList := ui.makeReport(w, req, []string{"svg"}, nil)
	if rpt == nil {
		return // error already reported
	}

	// Generate dot graph.
	g, config := report.GetDOT(rpt)
	legend := config.Labels
	config.Labels = nil
	dot := &bytes.Buffer{}
	graph.ComposeDot(dot, g, &graph.DotAttributes{}, config)

	defer cacher.Stop()

	result := &example.PodList{}

	if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
		if err := cacher.ready.wait(context.Background()); err != nil {
			t.Fatalf("unexpected error waiting for the cache to be ready")
		}
	}

	// Inject error to underlying layer and check if cacher is not bypassed.