kube client: use disk cache for CLI tools (istio#40386)

* kube client: use disk cache for CLI tools

When using istioctl on slow api-servers, things are super slow. The
issue is *every* call does API server discovery. This process is super
slow.

To workaround this slowness, k8s has caching. Today, we use an in-memory
cache. This is useless for CLI tools, which only run it once. Another
cache exists, which is used by kubectl, to cache to disk.

This PR introduces a disk based cache. This is enabled only for the
`NewExtendedClient`. To make it more explicit this should only be used
for CLI tools (and similar), the function is renamed to `NewCLIClient`.
Note all current cases of this function are correctly used in CLIs, so
this is just a renaming.

Additionally, add a new `pkg/lazy` package. This facilitates lazily
computing values, similar to `sync.Once` but more ergonomic and safer
(ability to retry, etc).

* Fix mapper to reset now that we are use cache

* rename and comment

* fmt

istioctl/cmd/analyze.go

@@ -130,7 +130,7 @@ func Analyze() *cobra.Command {
 
 			// check whether selected namespace exists.
 			if namespace != "" && useKube {
-				client, err := kube.NewExtendedClient(kube.BuildClientCmd(kubeconfig, configContext), "")
+				client, err := kube.NewClient(kube.BuildClientCmd(kubeconfig, configContext))
 				if err != nil {
 					return err
 				}

istioctl/cmd/authz.go

@@ -114,7 +114,7 @@ func getConfigDumpFromFile(filename string) (*configdump.Wrapper, error) {
 }
 
 func getConfigDumpFromPod(podName, podNamespace string) (*configdump.Wrapper, error) {
-	kubeClient, err := kube.NewExtendedClient(kube.BuildClientCmd(kubeconfig, configContext), "")
+	kubeClient, err := kube.NewCLIClient(kube.BuildClientCmd(kubeconfig, configContext), "")
 	if err != nil {
 		return nil, err
 	}

istioctl/cmd/dashboard.go

@@ -405,7 +405,7 @@ func skywalkingDashCmd() *cobra.Command {
 
 // portForward first tries to forward localhost:remotePort to podName:remotePort, falls back to dynamic local port
 func portForward(podName, namespace, flavor, urlFormat, localAddress string, remotePort int,
-	client kube.ExtendedClient, writer io.Writer, browser bool,
+	client kube.CLIClient, writer io.Writer, browser bool,
 ) error {
 	// port preference:
 	// - If --listenPort is specified, use it

istioctl/cmd/dashboard_test.go

@@ -125,10 +125,10 @@ func TestDashboard(t *testing.T) {
 	}
 }
 
-func mockExecClientDashboard(_, _, _ string) (kube.ExtendedClient, error) {
+func mockExecClientDashboard(_, _, _ string) (kube.CLIClient, error) {
 	return kube.MockClient{}, nil
 }
 
-func mockEnvoyClientDashboard(_, _ string) (kube.ExtendedClient, error) {
+func mockEnvoyClientDashboard(_, _ string) (kube.CLIClient, error) {
 	return kube.MockClient{}, nil
 }

istioctl/cmd/describe.go

@@ -881,7 +881,7 @@ func printVirtualService(writer io.Writer, vs *clientnetworking.VirtualService,
 	}
 }
 
-func printIngressInfo(writer io.Writer, matchingServices []v1.Service, podsLabels []k8s_labels.Set, kubeClient kubernetes.Interface, configClient istioclient.Interface, client kube.ExtendedClient) error { // nolint: lll
+func printIngressInfo(writer io.Writer, matchingServices []v1.Service, podsLabels []k8s_labels.Set, kubeClient kubernetes.Interface, configClient istioclient.Interface, client kube.CLIClient) error { // nolint: lll
 
 	pods, err := kubeClient.CoreV1().Pods(istioNamespace).List(context.TODO(), metav1.ListOptions{
 		LabelSelector: "istio=ingressgateway",
@@ -1124,7 +1124,7 @@ the configuration objects that affect that service.`,
 	return cmd
 }
 
-func describePodServices(writer io.Writer, kubeClient kube.ExtendedClient, configClient istioclient.Interface, pod *v1.Pod, matchingServices []v1.Service, podsLabels []k8s_labels.Set) error { // nolint: lll
+func describePodServices(writer io.Writer, kubeClient kube.CLIClient, configClient istioclient.Interface, pod *v1.Pod, matchingServices []v1.Service, podsLabels []k8s_labels.Set) error { // nolint: lll
 	byConfigDump, err := kubeClient.EnvoyDo(context.TODO(), pod.ObjectMeta.Name, pod.ObjectMeta.Namespace, "GET", "config_dump")
 	if err != nil {
 		if ignoreUnmeshed {
@@ -1216,7 +1216,7 @@ func containerReady(pod *v1.Pod, containerName string) (bool, error) {
 // describePeerAuthentication fetches all PeerAuthentication in workload and root namespace.
 // It lists the ones applied to the pod, and the current active mTLS mode.
 // When the client doesn't have access to root namespace, it will only show workload namespace Peerauthentications.
-func describePeerAuthentication(writer io.Writer, kubeClient kube.ExtendedClient, configClient istioclient.Interface, workloadNamespace string, podsLabels k8s_labels.Set) error { // nolint: lll
+func describePeerAuthentication(writer io.Writer, kubeClient kube.CLIClient, configClient istioclient.Interface, workloadNamespace string, podsLabels k8s_labels.Set) error { // nolint: lll
 	meshCfg, err := getMeshConfig(kubeClient)
 	if err != nil {
 		return fmt.Errorf("failed to fetch mesh config: %v", err)
@@ -1305,7 +1305,7 @@ func printPeerAuthentication(writer io.Writer, pa *v1beta1.PeerAuthentication) {
 	}
 }
 
-func getMeshConfig(kubeClient kube.ExtendedClient) (*meshconfig.MeshConfig, error) {
+func getMeshConfig(kubeClient kube.CLIClient) (*meshconfig.MeshConfig, error) {
 	rev := kubeClient.Revision()
 	meshConfigMapName := defaultMeshConfigMapName
 

istioctl/cmd/injector-list.go

@@ -133,7 +133,7 @@ func filterSystemNamespaces(nss []v1.Namespace) []v1.Namespace {
 	return filtered
 }
 
-func getNamespaces(ctx context.Context, client kube.ExtendedClient) ([]v1.Namespace, error) {
+func getNamespaces(ctx context.Context, client kube.CLIClient) ([]v1.Namespace, error) {
 	nslist, err := client.Kube().CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
 	if err != nil {
 		return []v1.Namespace{}, err
@@ -176,7 +176,7 @@ func printNS(writer io.Writer, namespaces []v1.Namespace, hooks []admit_v1.Mutat
 	return w.Flush()
 }
 
-func getWebhooks(ctx context.Context, client kube.ExtendedClient) ([]admit_v1.MutatingWebhookConfiguration, error) {
+func getWebhooks(ctx context.Context, client kube.CLIClient) ([]admit_v1.MutatingWebhookConfiguration, error) {
 	hooks, err := client.Kube().AdmissionregistrationV1().MutatingWebhookConfigurations().List(ctx, metav1.ListOptions{})
 	if err != nil {
 		return []admit_v1.MutatingWebhookConfiguration{}, err
@@ -256,7 +256,7 @@ func getMatchingNamespaces(hook *admit_v1.MutatingWebhookConfiguration, namespac
 	return retval
 }
 
-func getPods(ctx context.Context, client kube.ExtendedClient) (map[resource.Namespace][]v1.Pod, error) {
+func getPods(ctx context.Context, client kube.CLIClient) (map[resource.Namespace][]v1.Pod, error) {
 	retval := map[resource.Namespace][]v1.Pod{}
 	// All pods in all namespaces
 	pods, err := client.Kube().CoreV1().Pods("").List(ctx, metav1.ListOptions{})
@@ -275,7 +275,7 @@ func getPods(ctx context.Context, client kube.ExtendedClient) (map[resource.Name
 }
 
 // getInjectedImages() returns a map of revision->dockerimage
-func getInjectedImages(ctx context.Context, client kube.ExtendedClient) (map[string]string, error) {
+func getInjectedImages(ctx context.Context, client kube.CLIClient) (map[string]string, error) {
 	retval := map[string]string{}
 
 	// All configs in all namespaces that are Istio revisioned

istioctl/cmd/internal-debug.go

@@ -30,7 +30,7 @@ import (
 	"istio.io/istio/pkg/kube"
 )
 
-func HandlerForRetrieveDebugList(kubeClient kube.ExtendedClient,
+func HandlerForRetrieveDebugList(kubeClient kube.CLIClient,
 	centralOpts clioptions.CentralControlPlaneOptions,
 	writer io.Writer,
 ) (map[string]*xdsapi.DiscoveryResponse, error) {
@@ -51,7 +51,7 @@ func HandlerForRetrieveDebugList(kubeClient kube.ExtendedClient,
 	return xdsResponses, nil
 }
 
-func HandlerForDebugErrors(kubeClient kube.ExtendedClient,
+func HandlerForDebugErrors(kubeClient kube.CLIClient,
 	centralOpts *clioptions.CentralControlPlaneOptions,
 	writer io.Writer,
 	xdsResponses map[string]*xdsapi.DiscoveryResponse,

istioctl/cmd/kubeinject.go

@@ -66,7 +66,7 @@ const (
 )
 
 type ExternalInjector struct {
-	client          kube.ExtendedClient
+	client          kube.CLIClient
 	clientConfig    *admissionregistration.WebhookClientConfig
 	injectorAddress string
 }
@@ -345,7 +345,7 @@ func getInjectConfigFromConfigMap(kubeconfig, revision string) (inject.RawTempla
 
 func setUpExternalInjector(kubeconfig, revision, injectorAddress string) (*ExternalInjector, error) {
 	e := &ExternalInjector{}
-	client, err := kube.NewExtendedClient(kube.BuildClientCmd(kubeconfig, configContext), "")
+	client, err := kube.NewCLIClient(kube.BuildClientCmd(kubeconfig, configContext), "")
 	if err != nil {
 		return e, err
 	}

istioctl/cmd/metrics_test.go

@@ -36,7 +36,7 @@ type mockPromAPI struct {
 	cannedResponse map[string]prometheus_model.Value
 }
 
-func mockExecClientAuthNoPilot(_, _, _ string) (kube.ExtendedClient, error) {
+func mockExecClientAuthNoPilot(_, _, _ string) (kube.CLIClient, error) {
 	return &kube.MockClient{}, nil
 }
 
@@ -81,7 +81,7 @@ func TestMetrics(t *testing.T) {
 	}
 }
 
-func mockPortForwardClientAuthPrometheus(_, _, _ string) (kube.ExtendedClient, error) {
+func mockPortForwardClientAuthPrometheus(_, _, _ string) (kube.CLIClient, error) {
 	return &kube.MockClient{
 		DiscoverablePods: map[string]map[string]*v1.PodList{
 			"istio-system": {

istioctl/cmd/precheck.go

@@ -62,7 +62,7 @@ func preCheck() *cobra.Command {
   # Check only a single namespace
   istioctl x precheck --namespace default`,
 		RunE: func(cmd *cobra.Command, args []string) (err error) {
-			cli, err := kube.NewExtendedClient(kube.BuildClientCmd(kubeconfig, configContext), revision)
+			cli, err := kube.NewCLIClient(kube.BuildClientCmd(kubeconfig, configContext), revision)
 			if err != nil {
 				return err
 			}
@@ -106,7 +106,7 @@ See %s for more information about causes and resolutions.`, url.ConfigAnalysis)
 	return cmd
 }
 
-func checkControlPlane(cli kube.ExtendedClient) (diag.Messages, error) {
+func checkControlPlane(cli kube.CLIClient) (diag.Messages, error) {
 	msgs := diag.Messages{}
 
 	m, err := checkServerVersion(cli)
@@ -145,7 +145,7 @@ func checkControlPlane(cli kube.ExtendedClient) (diag.Messages, error) {
 	return msgs, nil
 }
 
-func checkInstallPermissions(cli kube.ExtendedClient) diag.Messages {
+func checkInstallPermissions(cli kube.CLIClient) diag.Messages {
 	Resources := []struct {
 		namespace string
 		group     string
@@ -222,7 +222,7 @@ func checkInstallPermissions(cli kube.ExtendedClient) diag.Messages {
 	return msgs
 }
 
-func checkCanCreateResources(c kube.ExtendedClient, namespace, group, version, name string) error {
+func checkCanCreateResources(c kube.CLIClient, namespace, group, version, name string) error {
 	s := &authorizationapi.SelfSubjectAccessReview{
 		Spec: authorizationapi.SelfSubjectAccessReviewSpec{
 			ResourceAttributes: &authorizationapi.ResourceAttributes{
@@ -249,7 +249,7 @@ func checkCanCreateResources(c kube.ExtendedClient, namespace, group, version, n
 	return nil
 }
 
-func checkServerVersion(cli kube.ExtendedClient) (diag.Messages, error) {
+func checkServerVersion(cli kube.CLIClient) (diag.Messages, error) {
 	v, err := cli.GetKubernetesVersion()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get the Kubernetes version: %v", err)
@@ -266,7 +266,7 @@ func checkServerVersion(cli kube.ExtendedClient) (diag.Messages, error) {
 	return nil, nil
 }
 
-func checkDataPlane(cli kube.ExtendedClient, namespace string) (diag.Messages, error) {
+func checkDataPlane(cli kube.CLIClient, namespace string) (diag.Messages, error) {
 	msgs := diag.Messages{}
 
 	m, err := checkListeners(cli, namespace)
@@ -280,7 +280,7 @@ func checkDataPlane(cli kube.ExtendedClient, namespace string) (diag.Messages, e
 	return msgs, nil
 }
 
-func checkListeners(cli kube.ExtendedClient, namespace string) (diag.Messages, error) {
+func checkListeners(cli kube.CLIClient, namespace string) (diag.Messages, error) {
 	pods, err := cli.Kube().CoreV1().Pods(namespace).List(context.Background(), metav1.ListOptions{
 		// Find all running pods
 		FieldSelector: "status.phase=Running",

istioctl/cmd/proxyconfig_test.go

@@ -210,8 +210,8 @@ func verifyExecTestOutput(t *testing.T, c execTestCase) {
 // mockClientExecFactoryGenerator generates a function with the same signature as
 // kubernetes.NewExecClient() that returns a mock client.
 // nolint: lll
-func mockClientExecFactoryGenerator(testResults map[string][]byte) func(kubeconfig, configContext string, _ string) (kube.ExtendedClient, error) {
-	outFactory := func(_, _ string, _ string) (kube.ExtendedClient, error) {
+func mockClientExecFactoryGenerator(testResults map[string][]byte) func(kubeconfig, configContext string, _ string) (kube.CLIClient, error) {
+	outFactory := func(_, _ string, _ string) (kube.CLIClient, error) {
 		return kube.MockClient{
 			Results: testResults,
 		}, nil
@@ -220,8 +220,8 @@ func mockClientExecFactoryGenerator(testResults map[string][]byte) func(kubeconf
 	return outFactory
 }
 
-func mockEnvoyClientFactoryGenerator(testResults map[string][]byte) func(kubeconfig, configContext string) (kube.ExtendedClient, error) {
-	outFactory := func(_, _ string) (kube.ExtendedClient, error) {
+func mockEnvoyClientFactoryGenerator(testResults map[string][]byte) func(kubeconfig, configContext string) (kube.CLIClient, error) {
+	outFactory := func(_, _ string) (kube.CLIClient, error) {
 		return kube.MockClient{
 			Results: testResults,
 		}, nil

istioctl/cmd/proxystatus.go

@@ -137,7 +137,7 @@ func readConfigFile(filename string) ([]byte, error) {
 	return data, nil
 }
 
-func newKubeClientWithRevision(kubeconfig, configContext string, revision string) (kube.ExtendedClient, error) {
+func newKubeClientWithRevision(kubeconfig, configContext string, revision string) (kube.CLIClient, error) {
 	rc, err := kube.DefaultRestConfig(kubeconfig, configContext, func(config *rest.Config) {
 		// We are running a one-off command locally, so we don't need to worry too much about rate limitting
 		// Bumping this up greatly decreases install time
@@ -147,10 +147,10 @@ func newKubeClientWithRevision(kubeconfig, configContext string, revision string
 	if err != nil {
 		return nil, err
 	}
-	return kube.NewExtendedClient(kube.NewClientConfigForRestConfig(rc), revision)
+	return kube.NewCLIClient(kube.NewClientConfigForRestConfig(rc), revision)
 }
 
-func newKubeClient(kubeconfig, configContext string) (kube.ExtendedClient, error) {
+func newKubeClient(kubeconfig, configContext string) (kube.CLIClient, error) {
 	return newKubeClientWithRevision(kubeconfig, configContext, "")
 }
 

istioctl/cmd/revision.go

@@ -540,7 +540,7 @@ func printSummaryTable(writer io.Writer, verbose bool, revisions map[string]*Rev
 	return tw.Flush()
 }
 
-func getAllIstioOperatorCRs(client kube.ExtendedClient) ([]*iopv1alpha1.IstioOperator, error) {
+func getAllIstioOperatorCRs(client kube.CLIClient) ([]*iopv1alpha1.IstioOperator, error) {
 	ucrs, err := client.Dynamic().Resource(istioOperatorGVR).
 		List(context.Background(), meta_v1.ListOptions{})
 	if err != nil {
@@ -654,7 +654,7 @@ func annotateWithNamespaceAndPodInfo(revDescription *RevisionDescription, revisi
 	return nil
 }
 
-func annotateWithGatewayInfo(revDescription *RevisionDescription, client kube.ExtendedClient) error {
+func annotateWithGatewayInfo(revDescription *RevisionDescription, client kube.CLIClient) error {
 	ingressPods, err := getPodsForComponent(client, "IngressGateways")
 	if err != nil {
 		return fmt.Errorf("error while fetching ingress gateway pods: %v", err)
@@ -668,7 +668,7 @@ func annotateWithGatewayInfo(revDescription *RevisionDescription, client kube.Ex
 	return nil
 }
 
-func annotateWithControlPlanePodInfo(revDescription *RevisionDescription, client kube.ExtendedClient) error {
+func annotateWithControlPlanePodInfo(revDescription *RevisionDescription, client kube.CLIClient) error {
 	controlPlanePods, err := getPodsForComponent(client, "Pilot")
 	if err != nil {
 		return fmt.Errorf("error while fetching control plane pods: %v", err)
@@ -980,7 +980,7 @@ func getEnabledComponents(iops *v1alpha1.IstioOperatorSpec) []string {
 	return enabledComponents
 }
 
-func getPodsForComponent(client kube.ExtendedClient, component string) ([]v1.Pod, error) {
+func getPodsForComponent(client kube.CLIClient, component string) ([]v1.Pod, error) {
 	return getPodsWithSelector(client, istioNamespace, &meta_v1.LabelSelector{
 		MatchLabels: map[string]string{
 			label.IoIstioRev.Name:        client.Revision(),
@@ -989,7 +989,7 @@ func getPodsForComponent(client kube.ExtendedClient, component string) ([]v1.Pod
 	})
 }
 
-func getPodsWithSelector(client kube.ExtendedClient, ns string, selector *meta_v1.LabelSelector) ([]v1.Pod, error) {
+func getPodsWithSelector(client kube.CLIClient, ns string, selector *meta_v1.LabelSelector) ([]v1.Pod, error) {
 	labelSelector, err := meta_v1.LabelSelectorAsSelector(selector)
 	if err != nil {
 		return []v1.Pod{}, err

istioctl/cmd/tag.go

@@ -266,7 +266,7 @@ revision tag before removing using the "istioctl tag list" command.
 }
 
 // setTag creates or modifies a revision tag.
-func setTag(ctx context.Context, kubeClient kube.ExtendedClient, tagName, revision, istioNS string, generate bool, w, stderr io.Writer) error {
+func setTag(ctx context.Context, kubeClient kube.CLIClient, tagName, revision, istioNS string, generate bool, w, stderr io.Writer) error {
 	opts := &tag.GenerateOptions{
 		Tag:                  tagName,
 		Revision:             revision,

istioctl/cmd/workload.go

@@ -290,7 +290,7 @@ func readWorkloadGroup(filename string, wg *clientv1alpha3.WorkloadGroup) error
 }
 
 // Creates all the relevant config for the given workload group and cluster
-func createConfig(kubeClient kube.ExtendedClient, wg *clientv1alpha3.WorkloadGroup, clusterID, ingressIP, internalIP,
+func createConfig(kubeClient kube.CLIClient, wg *clientv1alpha3.WorkloadGroup, clusterID, ingressIP, internalIP,
 	externalIP string, outputDir string, out io.Writer,
 ) error {
 	if err := os.MkdirAll(outputDir, filePerms); err != nil {
@@ -377,7 +377,7 @@ func createClusterEnv(wg *clientv1alpha3.WorkloadGroup, config *meshconfig.Proxy
 // Get and store the needed certificate and token. The certificate comes from the CA root cert, and
 // the token is generated by kubectl under the workload group's namespace and service account
 // TODO: Make the following accurate when using the Kubernetes certificate signer
-func createCertsTokens(kubeClient kube.ExtendedClient, wg *clientv1alpha3.WorkloadGroup, dir string, out io.Writer) error {
+func createCertsTokens(kubeClient kube.CLIClient, wg *clientv1alpha3.WorkloadGroup, dir string, out io.Writer) error {
 	rootCert, err := kubeClient.Kube().CoreV1().ConfigMaps(wg.Namespace).Get(context.Background(), controller.CACertNamespaceConfigMap, metav1.GetOptions{})
 	// errors if the requested configmap does not exist in the given namespace
 	if err != nil {
@@ -436,7 +436,7 @@ func createCertsTokens(kubeClient kube.ExtendedClient, wg *clientv1alpha3.Worklo
 	return nil
 }
 
-func createMeshConfig(kubeClient kube.ExtendedClient, wg *clientv1alpha3.WorkloadGroup, clusterID, dir, revision string) (*meshconfig.ProxyConfig, error) {
+func createMeshConfig(kubeClient kube.CLIClient, wg *clientv1alpha3.WorkloadGroup, clusterID, dir, revision string) (*meshconfig.ProxyConfig, error) {
 	istioCM := "istio"
 	// Case with multiple control planes
 	if isRevisioned(revision) {
@@ -555,7 +555,7 @@ func marshalWorkloadEntryPodPorts(p map[string]uint32) string {
 }
 
 // Retrieves the external IP of the ingress-gateway for the hosts file additions
-func createHosts(kubeClient kube.ExtendedClient, ingressIP, dir string, revision string) error {
+func createHosts(kubeClient kube.CLIClient, ingressIP, dir string, revision string) error {
 	// try to infer the ingress IP if the provided one is invalid
 	if validation.ValidateIPAddress(ingressIP) != nil {
 		p := strings.Split(ingressSvc, ".")
@@ -613,7 +613,7 @@ func mapToString(m map[string]string) string {
 }
 
 // extractClusterIDFromInjectionConfig can extract clusterID from injection configmap
-func extractClusterIDFromInjectionConfig(kubeClient kube.ExtendedClient) (string, error) {
+func extractClusterIDFromInjectionConfig(kubeClient kube.CLIClient) (string, error) {
 	injectionConfigMap := "istio-sidecar-injector"
 	// Case with multiple control planes
 	revision := kubeClient.Revision()

istioctl/cmd/workload_test.go

@@ -165,7 +165,7 @@ func TestWorkloadEntryConfigure(t *testing.T) {
 		}
 		t.Run(dir.Name(), func(t *testing.T) {
 			testdir := path.Join("testdata/vmconfig", dir.Name())
-			kubeClientWithRevision = func(_, _, _ string) (kube.ExtendedClient, error) {
+			kubeClientWithRevision = func(_, _, _ string) (kube.CLIClient, error) {
 				return &kube.MockClient{
 					RevisionValue: "rev-1",
 					Interface: fake.NewSimpleClientset(
@@ -259,7 +259,7 @@ func TestWorkloadEntryConfigureNilProxyMetadata(t *testing.T) {
 	testdir := "testdata/vmconfig-nil-proxy-metadata"
 	noClusterID := "failed to automatically determine the --clusterID"
 
-	kubeClientWithRevision = func(_, _, _ string) (kube.ExtendedClient, error) {
+	kubeClientWithRevision = func(_, _, _ string) (kube.CLIClient, error) {
 		return &kube.MockClient{
 			Interface: fake.NewSimpleClientset(
 				&v1.ServiceAccount{