Bump helm.sh/helm/v3 from 3.10.3 to 3.11.1 (#5528)

vendor/helm.sh/helm/v3/pkg/kube/client.go

@@ -17,12 +17,14 @@ limitations under the License.
 package kube // import "helm.sh/helm/v3/pkg/kube"
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
+	"reflect"
 	"strings"
 	"sync"
 	"time"
@@ -38,7 +40,9 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
 	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
@@ -47,6 +51,7 @@ import (
 	"k8s.io/cli-runtime/pkg/resource"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
 	cachetools "k8s.io/client-go/tools/cache"
 	watchtools "k8s.io/client-go/tools/watch"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
@@ -132,6 +137,141 @@ func (c *Client) Create(resources ResourceList) (*Result, error) {
 	return &Result{Created: resources}, nil
 }
 
+func transformRequests(req *rest.Request) {
+	tableParam := strings.Join([]string{
+		fmt.Sprintf("application/json;as=Table;v=%s;g=%s", metav1.SchemeGroupVersion.Version, metav1.GroupName),
+		fmt.Sprintf("application/json;as=Table;v=%s;g=%s", metav1beta1.SchemeGroupVersion.Version, metav1beta1.GroupName),
+		"application/json",
+	}, ",")
+	req.SetHeader("Accept", tableParam)
+
+	// if sorting, ensure we receive the full object in order to introspect its fields via jsonpath
+	req.Param("includeObject", "Object")
+}
+
+// Get retrieves the resource objects supplied. If related is set to true the
+// related pods are fetched as well. If the passed in resources are a table kind
+// the related resources will also be fetched as kind=table.
+func (c *Client) Get(resources ResourceList, related bool) (map[string][]runtime.Object, error) {
+	buf := new(bytes.Buffer)
+	objs := make(map[string][]runtime.Object)
+
+	podSelectors := []map[string]string{}
+	err := resources.Visit(func(info *resource.Info, err error) error {
+		if err != nil {
+			return err
+		}
+
+		gvk := info.ResourceMapping().GroupVersionKind
+		vk := gvk.Version + "/" + gvk.Kind
+		obj, err := getResource(info)
+		if err != nil {
+			fmt.Fprintf(buf, "Get resource %s failed, err:%v\n", info.Name, err)
+		} else {
+			objs[vk] = append(objs[vk], obj)
+
+			// Only fetch related pods if they are requested
+			if related {
+				// Discover if the existing object is a table. If it is, request
+				// the pods as Tables. Otherwise request them normally.
+				objGVK := obj.GetObjectKind().GroupVersionKind()
+				var isTable bool
+				if objGVK.Kind == "Table" {
+					isTable = true
+				}
+
+				objs, err = c.getSelectRelationPod(info, objs, isTable, &podSelectors)
+				if err != nil {
+					c.Log("Warning: get the relation pod is failed, err:%s", err.Error())
+				}
+			}
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return objs, nil
+}
+
+func (c *Client) getSelectRelationPod(info *resource.Info, objs map[string][]runtime.Object, table bool, podSelectors *[]map[string]string) (map[string][]runtime.Object, error) {
+	if info == nil {
+		return objs, nil
+	}
+	c.Log("get relation pod of object: %s/%s/%s", info.Namespace, info.Mapping.GroupVersionKind.Kind, info.Name)
+	selector, ok, _ := getSelectorFromObject(info.Object)
+	if !ok {
+		return objs, nil
+	}
+
+	for index := range *podSelectors {
+		if reflect.DeepEqual((*podSelectors)[index], selector) {
+			// check if pods for selectors are already added. This avoids duplicate printing of pods
+			return objs, nil
+		}
+	}
+
+	*podSelectors = append(*podSelectors, selector)
+
+	var infos []*resource.Info
+	var err error
+	if table {
+		infos, err = c.Factory.NewBuilder().
+			Unstructured().
+			ContinueOnError().
+			NamespaceParam(info.Namespace).
+			DefaultNamespace().
+			ResourceTypes("pods").
+			LabelSelector(labels.Set(selector).AsSelector().String()).
+			TransformRequests(transformRequests).
+			Do().Infos()
+		if err != nil {
+			return objs, err
+		}
+	} else {
+		infos, err = c.Factory.NewBuilder().
+			Unstructured().
+			ContinueOnError().
+			NamespaceParam(info.Namespace).
+			DefaultNamespace().
+			ResourceTypes("pods").
+			LabelSelector(labels.Set(selector).AsSelector().String()).
+			Do().Infos()
+		if err != nil {
+			return objs, err
+		}
+	}
+	vk := "v1/Pod(related)"
+
+	for _, info := range infos {
+		objs[vk] = append(objs[vk], info.Object)
+	}
+	return objs, nil
+}
+
+func getSelectorFromObject(obj runtime.Object) (map[string]string, bool, error) {
+	typed := obj.(*unstructured.Unstructured)
+	kind := typed.Object["kind"]
+	switch kind {
+	case "ReplicaSet", "Deployment", "StatefulSet", "DaemonSet", "Job":
+		return unstructured.NestedStringMap(typed.Object, "spec", "selector", "matchLabels")
+	case "ReplicationController":
+		return unstructured.NestedStringMap(typed.Object, "spec", "selector")
+	default:
+		return nil, false, nil
+	}
+}
+
+func getResource(info *resource.Info) (runtime.Object, error) {
+	obj, err := resource.NewHelper(info.Client, info.Mapping).Get(info.Namespace, info.Name)
+	if err != nil {
+		return nil, err
+	}
+	return obj, nil
+}
+
 // Wait waits up to the given timeout for the specified resources to be ready.
 func (c *Client) Wait(resources ResourceList, timeout time.Duration) error {
 	cs, err := c.getKubeClient()
@@ -215,6 +355,33 @@ func (c *Client) Build(reader io.Reader, validate bool) (ResourceList, error) {
 	return result, scrubValidationError(err)
 }
 
+// BuildTable validates for Kubernetes objects and returns unstructured infos.
+// The returned kind is a Table.
+func (c *Client) BuildTable(reader io.Reader, validate bool) (ResourceList, error) {
+	validationDirective := metav1.FieldValidationIgnore
+	if validate {
+		validationDirective = metav1.FieldValidationStrict
+	}
+
+	dynamicClient, err := c.Factory.DynamicClient()
+	if err != nil {
+		return nil, err
+	}
+
+	verifier := resource.NewQueryParamVerifier(dynamicClient, c.Factory.OpenAPIGetter(), resource.QueryParamFieldValidation)
+	schema, err := c.Factory.Validator(validationDirective, verifier)
+	if err != nil {
+		return nil, err
+	}
+	result, err := c.newBuilder().
+		Unstructured().
+		Schema(schema).
+		Stream(reader, "").
+		TransformRequests(transformRequests).
+		Do().Infos()
+	return result, scrubValidationError(err)
+}
+
 // Update takes the current list of objects and target list of objects and
 // creates resources that don't already exist, updates resources that have been
 // modified in the target configuration, and deletes resources from the current
@@ -308,16 +475,20 @@ func (c *Client) Delete(resources ResourceList) (*Result, []error) {
 	mtx := sync.Mutex{}
 	err := perform(resources, func(info *resource.Info) error {
 		c.Log("Starting delete for %q %s", info.Name, info.Mapping.GroupVersionKind.Kind)
-		if err := c.skipIfNotFound(deleteResource(info)); err != nil {
-			mtx.Lock()
-			defer mtx.Unlock()
-			// Collect the error and continue on
-			errs = append(errs, err)
-		} else {
+		err := deleteResource(info)
+		if err == nil || apierrors.IsNotFound(err) {
+			if err != nil {
+				c.Log("Ignoring delete failure for %q %s: %v", info.Name, info.Mapping.GroupVersionKind, err)
+			}
 			mtx.Lock()
 			defer mtx.Unlock()
 			res.Deleted = append(res.Deleted, info)
+			return nil
 		}
+		mtx.Lock()
+		defer mtx.Unlock()
+		// Collect the error and continue on
+		errs = append(errs, err)
 		return nil
 	})
 	if err != nil {
@@ -334,14 +505,6 @@ func (c *Client) Delete(resources ResourceList) (*Result, []error) {
 	return res, nil
 }
 
-func (c *Client) skipIfNotFound(err error) error {
-	if apierrors.IsNotFound(err) {
-		c.Log("%v", err)
-		return nil
-	}
-	return err
-}
-
 func (c *Client) watchTimeout(t time.Duration) func(*resource.Info) error {
 	return func(info *resource.Info) error {
 		return c.watchUntilReady(t, info)
@@ -356,10 +519,10 @@ func (c *Client) watchTimeout(t time.Duration) func(*resource.Info) error {
 // For most kinds, it checks to see if the resource is marked as Added or Modified
 // by the Kubernetes event stream. For some kinds, it does more:
 //
-// - Jobs: A job is marked "Ready" when it has successfully completed. This is
-//   ascertained by watching the Status fields in a job's output.
-// - Pods: A pod is marked "Ready" when it has successfully completed. This is
-//   ascertained by watching the status.phase field in a pod's output.
+//   - Jobs: A job is marked "Ready" when it has successfully completed. This is
+//     ascertained by watching the Status fields in a job's output.
+//   - Pods: A pod is marked "Ready" when it has successfully completed. This is
+//     ascertained by watching the status.phase field in a pod's output.
 //
 // Handling for other kinds will be added as necessary.
 func (c *Client) WatchUntilReady(resources ResourceList, timeout time.Duration) error {