if alreadyStarted {
		log.Debugf("cluster %q kube client started again", c.clusterID)
	} else {
		log.Infof("cluster %q kube client started", c.clusterID)
	}
}

func (c *client) GetKubernetesVersion() (*kubeVersion.Info, error) {
	return c.version.Get()
}

func (c *client) ClusterID() cluster.ID {
	return c.clusterID
}

		"DNS domain suffix")
	c.PersistentFlags().StringVar((*string)(&serverArgs.RegistryOptions.KubeOptions.ClusterID), "clusterID", features.ClusterName,
		"The ID of the cluster that this Istiod instance resides")
	c.PersistentFlags().StringToStringVar(&serverArgs.RegistryOptions.KubeOptions.ClusterAliases, "clusterAliases", map[string]string{},

		want     *core.Metadata
	}{
		{
			name: "all empty",
			metadata: &model.EndpointMetadata{
				TLSMode:      model.DisabledTLSModeLabel,
				Network:      "",
				WorkloadName: "",
				ClusterID:    "",
			},
			want: &core.Metadata{
				FilterMetadata: map[string]*structpb.Struct{
					IstioMetadataKey: {
						Fields: map[string]*structpb.Value{
							"workload": {

	cache := wasm.NewLocalFileCache(constants.IstioDataDir, ia.cfg.WASMOptions)
	proxy := &XdsProxy{
		istiodAddress:         ia.proxyConfig.DiscoveryAddress,
		istiodSAN:             ia.cfg.IstiodSAN,
		clusterID:             ia.secOpts.ClusterID,
		handlers:              map[string]ResponseHandler{},
		stopChan:              make(chan struct{}),

	// Create and run the controller.
	c1, ic := newTestServiceImportCache(t)

	// Create and run another controller.
	c2, _ := NewFakeControllerWithOptions(t, FakeControllerOptions{
		ClusterID: "test-cluster2",
	})

	c1.opts.MeshServiceController.AddRegistryAndRun(c2, c2.stop)

	ic.createKubeService(t, c1)
	ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)

		CAProviderName:                 caProviderEnv,
		PilotCertProvider:              features.PilotCertProvider,
		OutputKeyCertToDir:             outputKeyCertToDir,
		ProvCert:                       provCert,
		ClusterID:                      clusterIDVar.Get(),
		FileMountedCerts:               fileMountedCertsEnv,
		WorkloadNamespace:              PodNamespaceVar.Get(),
		ServiceAccount:                 serviceAccountVar.Get(),

		ConfigString: mustReadFile(t, "./testdata/ecds.yaml"),
	})

	ads := s.ConnectADS().WithType(v3.ExtensionConfigurationType)
	wantExtensionConfigName := "extension-config"
	md := model.NodeMetadata{
		ClusterID: constants.DefaultClusterName,
	}
	res := ads.RequestResponseAck(t, &discovery.DiscoveryRequest{
		Node: &core.Node{
			Id:       ads.ID,
			Metadata: md.ToStruct(),
		},

				Namespace: "foo",
			},
			want: kubetypes.InformerOptions{
				Namespace: "foo",
				Cluster:   c.ClusterID(),
			},
		},
		{
			name: "watch pods in the InformerWatchNamespace",
			gvr:  gvr.Pod,
			want: kubetypes.InformerOptions{
				Namespace: features.InformerWatchNamespace,
				Cluster:   c.ClusterID(),
			},
		},
		{
			name: "watch namespaces",
			gvr:  gvr.Namespace,

		return err
	}
	defer upstreamConn.Close()

	xds := discovery.NewAggregatedDiscoveryServiceClient(upstreamConn)
	ctx = metadata.AppendToOutgoingContext(context.Background(), "ClusterID", p.clusterID)
	for k, v := range p.xdsHeaders {
		ctx = metadata.AppendToOutgoingContext(ctx, k, v)
	}
	// We must propagate upstream termination to Envoy. This ensures that we resume the full XDS sequence on new connection

		Csr:              string(csrPEM),
		ValidityDuration: int64((time.Hour * 24 * 7).Seconds()),
	}
	rctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs("Authorization", "Bearer "+token, "ClusterID", constants.DefaultClusterName))
	resp, err := client.CreateCertificate(rctx, req)
	if err != nil {
		return Cert{}, fmt.Errorf("send CSR: %v", err)
	}
	certChain := []byte{}
	for _, c := range resp.CertChain {