func (c *Cluster) GetPodEventsWithSelector(ctx context.Context, namespace, selector string) (string, error) {
	podList, err := c.Kubectl.CoreV1().Pods(namespace).List(ctx,
		metav1.ListOptions{LabelSelector: selector})
	if err != nil {
		return "", err
	}
	if len(podList.Items) < 1 {
		return "", errors.Errorf("Couldn't find Pod with selector '%s' in namespace %s", selector, namespace)
	}
	podName := podList.Items[0].Name
	return c.GetPodEvents(ctx, namespace, podName)
}
GetPodEventsWithSelector tries to find a pod using the provided selector and
namespace. If found it returns the events on that Pod. If not found it returns
an error.
An equivalent kubectl command would look like this
(label selector being "app.kubernetes.io/name=container-registry"):
kubectl get event --namespace my-namespace \
--field-selector involvedObject.name=$( \
kubectl get pods -o=jsonpath='{.items[0].metadata.name}' --selector=app.kubernetes.io/name=container-registry -n my-namespace)
package kubernetes
import (
	"bytes"
	"context"
	"fmt"
	"io"
	"strings"
	"time"
	"github.com/pkg/errors"
	kubeconfig "github.com/epinio/epinio/helpers/kubernetes/config"
	generic "github.com/epinio/epinio/helpers/kubernetes/platform/generic"
	ibm "github.com/epinio/epinio/helpers/kubernetes/platform/ibm"
	k3s "github.com/epinio/epinio/helpers/kubernetes/platform/k3s"
	kind "github.com/epinio/epinio/helpers/kubernetes/platform/kind"
	minikube "github.com/epinio/epinio/helpers/kubernetes/platform/minikube"
	"github.com/epinio/epinio/helpers/termui"
	tekton "github.com/tektoncd/pipeline/pkg/client/clientset/versioned"
	tektonv1beta1 "github.com/tektoncd/pipeline/pkg/client/clientset/versioned/typed/pipeline/v1beta1"
	appsv1 "k8s.io/api/apps/v1"
	apibatchv1 "k8s.io/api/batch/v1"
	v1 "k8s.io/api/core/v1"
	networkingv1 "k8s.io/api/networking/v1"
	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
	apiextensions "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	typedbatchv1 "k8s.io/client-go/kubernetes/typed/batch/v1"
	restclient "k8s.io/client-go/rest"
	"k8s.io/client-go/tools/remotecommand"
	_ "k8s.io/client-go/plugin/pkg/client/auth/oidc"
)
const (
	APISGroupName = "epinio.suse.org"
)
var (
	EpinioDeploymentLabelKey   = fmt.Sprintf("%s/%s", APISGroupName, "deployment")
	EpinioDeploymentLabelValue = "true"
	EpinioOrgLabelKey          = "app.kubernetes.io/component"
	EpinioOrgLabelValue        = "epinio-namespace"
	EpinioAPISecretLabelKey    = fmt.Sprintf("%s/%s", APISGroupName, "api-user-credentials")
	EpinioAPISecretLabelValue  = "true"
)
var clusterMemo *Cluster
type Platform interface {
	Detect(context.Context, *kubernetes.Clientset) bool
	Describe() string
	String() string
	Load(context.Context, *kubernetes.Clientset) error
	ExternalIPs() []string
}
var SupportedPlatforms = []Platform{
	kind.NewPlatform(),
	k3s.NewPlatform(),
	ibm.NewPlatform(),
	minikube.NewPlatform(),
}
type Cluster struct {
	Kubectl    *kubernetes.Clientset
	RestConfig *restclient.Config
	platform   Platform
}
func GetCluster(ctx context.Context) (*Cluster, error) {
	if clusterMemo != nil {
		return clusterMemo, nil
	}
	c := &Cluster{}
	restConfig, err := kubeconfig.KubeConfig()
	if err != nil {
		return nil, err
	}
	config := restclient.CopyConfig(restConfig)
	config.WarningHandler = restclient.NoWarnings{}
	c.RestConfig = restConfig
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, err
	}
	c.Kubectl = clientset
	c.detectPlatform(ctx)
	if c.platform == nil {
		c.platform = generic.NewPlatform()
	}
	err = c.platform.Load(ctx, clientset)
	if err != nil {
		return nil, err
	}
	clusterMemo = c
	return clusterMemo, nil
}
func (c *Cluster) GetPlatform() Platform {
	return c.platform
}
func (c *Cluster) detectPlatform(ctx context.Context) {
	for _, p := range SupportedPlatforms {
		if p.Detect(ctx, c.Kubectl) {
			c.platform = p
			return
		}
	}
}
func (c *Cluster) ClientApp() (dynamic.NamespaceableResourceInterface, error) {
	cs, err := dynamic.NewForConfig(c.RestConfig)
	if err != nil {
		return nil, err
	}
	gvr := schema.GroupVersionResource{
		Group:    "application.epinio.io",
		Version:  "v1",
		Resource: "apps",
	}
	return cs.Resource(gvr), nil
}
func (c *Cluster) ClientCertManager() (dynamic.NamespaceableResourceInterface, error) {
	gvr := schema.GroupVersionResource{
		Group:    "cert-manager.io",
		Version:  "v1",
		Resource: "clusterissuers",
	}
	dynamicClient, err := dynamic.NewForConfig(c.RestConfig)
	if err != nil {
		return nil, err
	}
	return dynamicClient.Resource(gvr), nil
}
func (c *Cluster) ClientCertificate() (dynamic.NamespaceableResourceInterface, error) {
	gvr := schema.GroupVersionResource{
		Group:    "cert-manager.io",
		Version:  "v1",
		Resource: "certificates",
	}
	dynamicClient, err := dynamic.NewForConfig(c.RestConfig)
	if err != nil {
		return nil, err
	}
	return dynamicClient.Resource(gvr), nil
}
func (c *Cluster) ClientTekton() (tektonv1beta1.TektonV1beta1Interface, error) {
	cs, err := tekton.NewForConfig(c.RestConfig)
	if err != nil {
		return nil, err
	}
	return cs.TektonV1beta1(), nil
}
func (c *Cluster) IsPodRunning(ctx context.Context, podName, namespace string) wait.ConditionFunc {
	return func() (bool, error) {
		pod, err := c.Kubectl.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		for _, condition := range pod.Status.Conditions {
			if condition.Type == v1.PodReady && condition.Status == v1.ConditionTrue {
				return true, nil
			}
		}
		return false, nil
	}
}
func (c *Cluster) IsJobCompleted(ctx context.Context, client *typedbatchv1.BatchV1Client, jobName, namespace string) wait.ConditionFunc {
	return func() (bool, error) {
		job, err := client.Jobs(namespace).Get(ctx, jobName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		for _, condition := range job.Status.Conditions {
			if condition.Type == apibatchv1.JobComplete && condition.Status == v1.ConditionTrue {
				return true, nil
			}
		}
		return false, nil
	}
}
func (c *Cluster) PodExists(ctx context.Context, namespace, selector string) wait.ConditionFunc {
	return func() (bool, error) {
		podList, err := c.ListPods(ctx, namespace, selector)
		if err != nil {
			return false, err
		}
		if len(podList.Items) == 0 {
			return false, nil
		}
		return true, nil
	}
}
func (c *Cluster) DeploymentExists(ctx context.Context, namespace, deploymentName string) wait.ConditionFunc {
	return func() (bool, error) {
		_, err := c.Kubectl.AppsV1().Deployments(namespace).Get(ctx, deploymentName, metav1.GetOptions{})
		if err != nil {
			if apierrors.IsNotFound(err) {
				return false, nil
			}
			return false, err
		}
		return true, nil
	}
}
func (c *Cluster) NamespaceDoesNotExist(ctx context.Context, namespaceName string) wait.ConditionFunc {
	return func() (bool, error) {
		exists, err := c.NamespaceExists(ctx, namespaceName)
		return !exists, err
	}
}
func (c *Cluster) PodDoesNotExist(ctx context.Context, namespace, selector string) wait.ConditionFunc {
	return func() (bool, error) {
		podList, err := c.ListPods(ctx, namespace, selector)
		if err != nil {
			return true, nil
		}
		if len(podList.Items) > 0 {
			return false, nil
		}
		return true, nil
	}
}
func (c *Cluster) WaitForCRD(ctx context.Context, ui *termui.UI, CRDName string, timeout time.Duration) error {
	s := ui.Progressf("Waiting for CRD %s to be ready to use", CRDName)
	defer s.Stop()
	clientset, err := apiextensions.NewForConfig(c.RestConfig)
	if err != nil {
		return err
	}
	err = wait.PollImmediate(time.Second, timeout, func() (bool, error) {
		_, err = clientset.ApiextensionsV1().CustomResourceDefinitions().Get(ctx, CRDName, metav1.GetOptions{})
		if err != nil {
			if apierrors.IsNotFound(err) {
				return false, nil
			}
			return false, err
		}
		return true, nil
	})
	if err != nil {
		return err
	}
	return wait.PollImmediate(time.Second, timeout, func() (bool, error) {
		crd, err := clientset.ApiextensionsV1().CustomResourceDefinitions().Get(ctx, CRDName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		for _, cond := range crd.Status.Conditions {
			if cond.Type == apiextensionsv1.Established && cond.Status == apiextensionsv1.ConditionTrue {
				return true, nil
			}
		}
		return false, nil
	})
}
func (c *Cluster) WaitForSecret(ctx context.Context, namespace, secretName string, timeout time.Duration) (*v1.Secret, error) {
	var secret *v1.Secret
	waitErr := wait.PollImmediate(time.Second, timeout, func() (bool, error) {
		var err error
		secret, err = c.GetSecret(ctx, namespace, secretName)
		if err != nil {
			if apierrors.IsNotFound(err) {
				return false, nil
			}
			return false, err
		}
		return true, nil
	})
	return secret, waitErr
}
func (c *Cluster) WaitForPodRunning(ctx context.Context, namespace, podName string, timeout time.Duration) error {
	return wait.PollImmediate(time.Second, timeout, c.IsPodRunning(ctx, podName, namespace))
}
func (c *Cluster) WaitForJobCompleted(ctx context.Context, namespace, jobName string, timeout time.Duration) error {
	client, err := typedbatchv1.NewForConfig(c.RestConfig)
	if err != nil {
		return err
	}
	return wait.PollImmediate(time.Second, timeout, c.IsJobCompleted(ctx, client, jobName, namespace))
}
func (c *Cluster) IsDeploymentCompleted(ctx context.Context, deploymentName, namespace string) wait.ConditionFunc {
	return func() (bool, error) {
		deployment, err := c.Kubectl.AppsV1().Deployments(namespace).Get(ctx,
			deploymentName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		for _, condition := range deployment.Status.Conditions {
			if condition.Type == appsv1.DeploymentAvailable && condition.Status == v1.ConditionTrue {
				return true, nil
			}
		}
		return false, nil
	}
}
func (c *Cluster) WaitForDeploymentCompleted(ctx context.Context, ui *termui.UI, namespace, deploymentName string, timeout time.Duration) error {
	if ui != nil {
		s := ui.Progressf("Waiting for deployment %s in %s to be ready", deploymentName, namespace)
		defer s.Stop()
	}
	return wait.PollImmediate(time.Second, timeout, c.IsDeploymentCompleted(ctx, deploymentName, namespace))
}
func (c *Cluster) ListPods(ctx context.Context, namespace, selector string) (*v1.PodList, error) {
	listOptions := metav1.ListOptions{}
	if len(selector) > 0 {
		listOptions.LabelSelector = selector
	}
	podList, err := c.Kubectl.CoreV1().Pods(namespace).List(ctx, listOptions)
	if err != nil {
		return nil, err
	}
	return podList, nil
}
func (c *Cluster) WaitForNamespaceMissing(ctx context.Context, ui *termui.UI, namespace string, timeout time.Duration) error {
	if ui != nil {
		s := ui.Progressf("Waiting for namespace %s to be deleted", namespace)
		defer s.Stop()
	}
	return wait.PollImmediate(time.Second, timeout, c.NamespaceDoesNotExist(ctx, namespace))
}
func (c *Cluster) WaitForNamespace(ctx context.Context, ui *termui.UI, namespace string, timeout time.Duration) error {
	s := ui.Progressf("Waiting for namespace %s to be appear", namespace)
	defer s.Stop()
	return wait.PollImmediate(time.Second, timeout, func() (bool, error) {
		exists, err := c.NamespaceExists(ctx, namespace)
		return exists, err
	})
}
func (c *Cluster) WaitForPodBySelectorMissing(ctx context.Context, ui *termui.UI, namespace, selector string, timeout time.Duration) error {
	if ui != nil {
		s := ui.Progressf("Removing %s in %s", selector, namespace)
		defer s.Stop()
	}
	return wait.PollImmediate(time.Second, timeout, c.PodDoesNotExist(ctx, namespace, selector))
}
func (c *Cluster) WaitUntilDeploymentExists(ctx context.Context, ui *termui.UI, namespace, deploymentName string, timeout time.Duration) error {
	s := ui.Progressf("Waiting for deployment %s in %s to appear", deploymentName, namespace)
	defer s.Stop()
	return wait.PollImmediate(time.Second, timeout, c.DeploymentExists(ctx, namespace, deploymentName))
}
func (c *Cluster) WaitUntilServiceHasLoadBalancer(ctx context.Context, ui *termui.UI, namespace, serviceName string, timeout time.Duration) error {
	s := ui.Progressf("Waiting for service %s in %s to have a Load Balancer", serviceName, namespace)
	defer s.Stop()
	return wait.PollImmediate(time.Second, timeout, func() (bool, error) {
		service, err := c.Kubectl.CoreV1().Services(namespace).Get(ctx, serviceName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		if len(service.Status.LoadBalancer.Ingress) == 0 {
			return false, nil
		}
		return true, nil
	})
}
func (c *Cluster) WaitUntilPodBySelectorExist(ctx context.Context, ui *termui.UI, namespace, selector string, timeout time.Duration) error {
	s := ui.Progressf("Creating %s in %s", selector, namespace)
	defer s.Stop()
	return wait.PollImmediate(time.Second, timeout, c.PodExists(ctx, namespace, selector))
}
func (c *Cluster) WaitForPodBySelectorRunning(ctx context.Context, ui *termui.UI, namespace, selector string, timeout time.Duration) error {
	s := ui.Progressf("Starting %s in %s", selector, namespace)
	defer s.Stop()
	podList, err := c.ListPods(ctx, namespace, selector)
	if err != nil {
		return errors.Wrapf(err, "failed listingpods with selector %s", selector)
	}
	if len(podList.Items) == 0 {
		return fmt.Errorf("no pods in %s with selector %s", namespace, selector)
	}
	for _, pod := range podList.Items {
		s.ChangeMessagef("  Starting pod %s in %s", pod.Name, namespace)
		if err := c.WaitForPodRunning(ctx, namespace, pod.Name, timeout); err != nil {
			events, err2 := c.GetPodEvents(ctx, namespace, pod.Name)
			if err2 != nil {
				return errors.Wrap(err, err2.Error())
			}
			return fmt.Errorf("Failed waiting for %s: %s\nPod Events: \n%s", pod.Name, err.Error(), events)
		}
	}
	return nil
}