func (s *Server) initServiceControllers(args *PilotArgs) error {
	serviceControllers := s.ServiceController()

	s.serviceEntryController = serviceentry.NewController(
		s.configController, s.XDSServer,
		s.environment.Watcher,
		serviceentry.WithClusterID(s.clusterID),
	)
	serviceControllers.AddRegistry(s.serviceEntryController)

	registered := sets.New[provider.ID]()
	for _, r := range args.RegistryOptions.Registries {

	}

	w, err := store.watcher.Watch(ctx, "/abc", int64(102), requestOpts)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Stop()

	actualEvent := <-w.ResultChan()
	if actualEvent.Type != watch.Error {
		t.Fatalf("Unexpected type of the event: %v, expected: %v", actualEvent.Type, watch.Error)
	}
	actualErr, ok := actualEvent.Object.(*metav1.Status)
	if !ok {

	wg.Wait()

	if watchErr != nil {
		t.Fatal(watchErr)
	}

	// wait out the expiration period and pop expired watchers
	time.Sleep(2 * time.Second)
	cacher.Lock()
	defer cacher.Unlock()
	cacher.bookmarkWatchers.popExpiredWatchersThreadUnsafe()
	if len(cacher.bookmarkWatchers.watchersBuckets) != 0 {
		numWatchers := 0
		for bucketID, v := range cacher.bookmarkWatchers.watchersBuckets {

	if !cfg.Enabled {
		return nil, nil
	}
	cli, err := clientv3.New(cfg.Config)
	if err != nil {
		return nil, err
	}
	cli.KV = namespace.NewKV(cli.KV, cfg.PathPrefix)
	cli.Watcher = namespace.NewWatcher(cli.Watcher, cfg.PathPrefix)
	cli.Lease = namespace.NewLease(cli.Lease, cfg.PathPrefix)
	return cli, nil
}

func parseEndpoints(endpoints string) ([]string, bool, error) {

	return &c, nil
}

// WebhookParameters configures parameters for the sidecar injection
// webhook.
type WebhookParameters struct {
	// Watcher watches the sidecar injection configuration.
	Watcher Watcher

	// Port is the webhook port, e.g. typically 443 for https.
	// This is mainly used for tests. Webhook runs on the port started by Istiod.
	Port int

	Env *model.Environment

	close(conn.initialized)
}

// ConnectionContext is used by the RPC event loop to respond to requests and pushes.
type ConnectionContext interface {
	XdsConnection() *Connection
	Watcher() Watcher
	// Initialize checks the first request.
	Initialize(node *core.Node) error
	// Close discards the connection.
	Close()
	// Process responds to a discovery request.

}

func newLeaderElection(namespace, name, electionID, revision string, perRevision bool, remote bool, client kube.Client) *LeaderElection {
	var watcher revisions.DefaultWatcher
	if features.EnableLeaderElection {
		watcher = revisions.NewDefaultWatcher(client, revision)
	}
	if name == "" {
		hn, _ := os.Hostname()
		name = fmt.Sprintf("unknown-%s", hn)
	}
	if perRevision && revision != "" {

}

func (c *Context) Watcher() xds.Watcher {
	return c.w
}

func (w *Watch) DeleteWatchedResource(string) {
	w.Lock()
	defer w.Unlock()
	w.watch = nil
}

func (w *Watch) GetWatchedResource(string) *xds.WatchedResource {
	w.Lock()
	defer w.Unlock()
	return w.watch
}

func (w *Watch) NewWatchedResource(typeURL string, names []string) {
	w.Lock()

	return append(certPEM, caCert...), caCert, nil
}

// Return signed CSR through a watcher. If no CSR is read, return nil.
func readSignedCsr(client clientset.Interface, csr string, watchTimeout time.Duration) ([]byte, error) {
	selector := fields.OneTermEqualSelector("metadata.name", csr).String()
	// Setup a List+Watch, like informers do
	// A simple Watch will fail if the cert is signed too quickly

// shouldn't we be doing this for both validating and mutating webhooks...?
func (c *Controller) startCaBundleWatcher(stop <-chan struct{}) {
	if c.o.CABundleWatcher == nil {
		return
	}
	id, watchCh := c.o.CABundleWatcher.AddWatcher()
	defer c.o.CABundleWatcher.RemoveWatcher(id)

	for {
		select {
		case <-watchCh:
			c.syncAll()
		case <-stop: