// Copyright 2017 Istio Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package util

import (
	"bufio"
	"bytes"
	"context"
	"errors"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"text/template"
	"time"

	"golang.org/x/net/context/ctxhttp"
	"golang.org/x/sync/errgroup"
	"istio.io/pkg/log"
)

const (
	podFailedGet = "Failed_Get"
	// The index of STATUS field in kubectl CLI output.
	statusField            = 2
	defaultClusterSubnet   = "24"
	defaultClusterSubnetv6 = "128"

	// NodePortServiceType NodePort type of Kubernetes Service
	NodePortServiceType = "NodePort"

	// LoadBalancerServiceType LoadBalancer type of Kubernetes Service
	LoadBalancerServiceType = "LoadBalancer"

	ingressHTTPServicePort = "80"
)

var (
	logDumpResources = []string{
		"pod",
		"service",
		"ingress",
	}
)

// PodInfo contains pod's information such as name and IP address
type PodInfo struct {
	// Name is the pod's name
	Name string
	// IPAddr is the pod's IP
	IPAddr string
}

// Fill complete a template with given values and generate a new output file
func Fill(outFile, inFile string, values interface{}) error {
	tmpl, err := template.ParseFiles(inFile)
	if err != nil {
		return err
	}

	var filled bytes.Buffer
	w := bufio.NewWriter(&filled)
	if err := tmpl.Execute(w, values); err != nil {
		return err
	}

	if err := w.Flush(); err != nil {
		return err
	}

	if err := ioutil.WriteFile(outFile, filled.Bytes(), 0644); err != nil {
		return err
	}
	log.Infof("Created %s from template %s", outFile, inFile)
	return nil
}

// CreateAndFill fills in the given yaml template with the values and generates a temp file for the completed yaml.
func CreateAndFill(outDir, templateFile string, values interface{}) (string, error) {
	outFile, err := CreateTempfile(outDir, filepath.Base(templateFile), "yaml")
	if err != nil {
		log.Errorf("Failed to generate yaml %s: %v", templateFile, err)
		return "", err
	}
	if err := Fill(outFile, templateFile, values); err != nil {
		log.Errorf("Failed to generate yaml for template %s: %v", templateFile, err)
		return "", err
	}
	return outFile, nil
}

// CreateNamespace create a kubernetes namespace
func CreateNamespace(n string, kubeconfig string) error {
	if _, err := ShellMuteOutput("kubectl create namespace %s --kubeconfig=%s", n, kubeconfig); err != nil {
		if !strings.Contains(err.Error(), "AlreadyExists") {
			return err
		}
	}
	log.Infof("namespace %s created\n", n)
	return nil
}

// DeleteNamespace delete a kubernetes namespace
func DeleteNamespace(n string, kubeconfig string) error {
	if _, err := Shell("kubectl delete namespace %s --kubeconfig=%s", n, kubeconfig); err != nil {
		if !strings.Contains(err.Error(), "not found") {
			return err
		}
	}
	log.Infof("namespace %s deleted\n", n)
	return nil
}

// DeleteDeployment deletes deployment from the specified namespace
func DeleteDeployment(d string, n string, kubeconfig string) error {
	_, err := Shell("kubectl delete deployment %s -n %s --kubeconfig=%s", d, n, kubeconfig)
	return err
}

// LabelNamespace will add a label to the kubernetes namespace
func LabelNamespace(n, label, kubeconfig string) error {
	if _, err := Shell("kubectl label namespace %s %s --overwrite=true --kubeconfig=%s", n, label, kubeconfig); err != nil {
		if !strings.Contains(err.Error(), "AlreadyExists") {
			return err
		}
	}
	log.Infof("label %s added to namespace %s", label, n)
	return nil
}

// NamespaceDeleted check if a kubernete namespace is deleted
func NamespaceDeleted(n string, kubeconfig string) (bool, error) {
	output, err := ShellSilent("kubectl get namespace %s -o name --kubeconfig=%s", n, kubeconfig)
	if strings.Contains(output, "NotFound") {
		return true, nil
	}
	return false, err
}

// ValidatingWebhookConfigurationExists check if a kubernetes ValidatingWebhookConfiguration is deleted
func ValidatingWebhookConfigurationExists(name string, kubeconfig string) bool {
	output, _ := ShellSilent("kubectl get validatingwebhookconfiguration %s -o name --kubeconfig=%s", name, kubeconfig)
	return !strings.Contains(output, "NotFound")
}

// KubeApplyContents kubectl apply from contents
func KubeApplyContents(namespace, yamlContents string, kubeconfig string) error {
	tmpfile, err := WriteTempfile(os.TempDir(), "kubeapply", ".yaml", yamlContents)
	if err != nil {
		return err
	}
	defer removeFile(tmpfile)
	return KubeApply(namespace, tmpfile, kubeconfig)
}

func kubeCommand(subCommand, namespace, yamlFileName string, kubeconfig string) string {
	if namespace == "" {
		if yamlFileName == "" {
			return fmt.Sprintf("kubectl %s --kubeconfig=%s", subCommand, kubeconfig)
		}
		return fmt.Sprintf("kubectl %s -f %s --kubeconfig=%s", subCommand, yamlFileName, kubeconfig)
	}
	if yamlFileName == "" {
		return fmt.Sprintf("kubectl %s -n %s --kubeconfig=%s", subCommand, namespace, kubeconfig)
	}
	return fmt.Sprintf("kubectl %s -n %s -f %s --kubeconfig=%s", subCommand, namespace, yamlFileName, kubeconfig)
}

// KubeApply kubectl apply from file
func KubeApply(namespace, yamlFileName string, kubeconfig string) error {
	_, err := Shell(kubeCommand("apply", namespace, yamlFileName, kubeconfig))
	return err
}

// KubeCommand executes the given kubectl command with the given yaml file
func KubeCommand(command, namespace, yamlFileName string, kubeconfig string) (string, error) {
	result, err := Shell(kubeCommand(command, namespace, yamlFileName, kubeconfig))
	return result, err
}

// KubeGetYaml kubectl get yaml content for given resource.
func KubeGetYaml(namespace, resource, name string, kubeconfig string) (string, error) {
	if namespace == "" {
		namespace = "default"
	}
	cmd := fmt.Sprintf("kubectl get %s %s -n %s -o yaml --kubeconfig=%s", resource, name, namespace, kubeconfig)

	return Shell(cmd)
}

// KubeApplyContentSilent kubectl apply from contents silently
func KubeApplyContentSilent(namespace, yamlContents string, kubeconfig string) error {
	tmpfile, err := WriteTempfile(os.TempDir(), "kubeapply", ".yaml", yamlContents)
	if err != nil {
		return err
	}
	defer removeFile(tmpfile)
	return KubeApplySilent(namespace, tmpfile, kubeconfig)
}

// KubeApplySilent kubectl apply from file silently
func KubeApplySilent(namespace, yamlFileName string, kubeconfig string) error {
	_, err := ShellSilent(kubeCommand("apply", namespace, yamlFileName, kubeconfig))
	return err
}

// KubeScale kubectl scale a pod specified using typeName
func KubeScale(namespace, typeName string, replicaCount int, kubeconfig string) error {
	kubecommand := fmt.Sprintf("kubectl scale -n %s --replicas=%d %s --kubeconfig=%s", namespace, replicaCount,
		typeName, kubeconfig)
	_, err := Shell(kubecommand)
	return err
}

// KubeDeleteContents kubectl apply from contents
func KubeDeleteContents(namespace, yamlContents string, kubeconfig string) error {
	tmpfile, err := WriteTempfile(os.TempDir(), "kubedelete", ".yaml", yamlContents)
	if err != nil {
		return err
	}
	defer removeFile(tmpfile)
	return KubeDelete(namespace, tmpfile, kubeconfig)
}

func removeFile(path string) {
	err := os.RemoveAll(path)
	if err != nil {
		log.Errorf("Unable to remove %s: %v", path, err)
	}
}

// KubeDelete kubectl delete from file
func KubeDelete(namespace, yamlFileName string, kubeconfig string) error {
	_, err := Shell(kubeCommand("delete", namespace, yamlFileName, kubeconfig))
	return err
}

// GetKubeMasterIP returns the IP address of the kubernetes master service.
func GetKubeMasterIP(kubeconfig string) (string, error) {
	return ShellSilent("kubectl get svc kubernetes -n default -o jsonpath='{.spec.clusterIP}' --kubeconfig=%s", kubeconfig)
}

// GetClusterSubnet returns the subnet (in CIDR form, e.g. "24") for the nodes in the cluster.
func GetClusterSubnet(kubeconfig string) (string, error) {
	cidr, err := ShellSilent("kubectl get nodes -o jsonpath='{.items[0].spec.podCIDR}' --kubeconfig=%s", kubeconfig)
	if err != nil {
		// This command should never fail. If the field isn't found, it will just return and empty string.
		return "", err
	}
	parts := strings.Split(cidr, "/")
	if len(parts) != 2 {
		ip, _ := GetKubeMasterIP(kubeconfig)
		addr := net.ParseIP(ip)
		if addr == nil {
			return "", fmt.Errorf("unable to determine the kubernetes service IP and cluster subnet")
		}
		if addr.To4() != nil {
			// TODO(nmittler): Need a way to get the subnet on minikube. For now, just return a default value.
			log.Info("unable to identify cluster subnet. running on minikube?")
			return defaultClusterSubnet, nil
		}
		log.Info("unable to identify IPv6 cluster subnet")
		return defaultClusterSubnetv6, nil
	}
	return parts[1], nil
}

func getRetrier(serviceType string) Retrier {
	baseDelay := 1 * time.Second
	maxDelay := 1 * time.Second
	retries := 300 // ~5 minutes

	if serviceType == NodePortServiceType {
		baseDelay = 5 * time.Second
		maxDelay = 5 * time.Second
		retries = 20
	}

	return Retrier{
		BaseDelay: baseDelay,
		MaxDelay:  maxDelay,
		Retries:   retries,
	}
}

// GetServiceIP get the service clusterIP.
func GetServiceIP(serviceName, namespace, kubeconfig string) (string, error) {
	res, err := Shell("kubectl -n %s get svc %s -o go-template='{{$.spec.clusterIP}}' --kubeconfig=%s",
		namespace, serviceName, kubeconfig)
	return strings.TrimSpace(res), err
}

// GetIngress get istio ingress ip and port. Could relate to either Istio Ingress or to
// Istio Ingress Gateway, by serviceName and podLabel. Handles two cases: when the Ingress/Ingress Gateway
// Kubernetes Service is a LoadBalancer or NodePort (for tests within the  cluster, including for minikube)
func GetIngress(serviceName, podLabel, namespace, kubeconfig string, serviceType string,
	sanityCheck bool) (string, error) {

	retry := getRetrier(serviceType)
	var ingress string
	retryFn := func(_ context.Context, i int) error {
		var err error

		if serviceType == NodePortServiceType {
			ingress, err = getServiceNodePort(serviceName, podLabel, namespace, kubeconfig)
		} else if serviceType == LoadBalancerServiceType {
			ingress, err = getServiceLoadBalancer(serviceName, namespace, kubeconfig)
		} else {
			return fmt.Errorf("unknown service type: %s", serviceType)
		}

		if err != nil {
			return err
		}
		log.Infof("Istio ingress: %s", ingress)
		return nil
	}

	ctx := context.Background()

	log.Infof("Waiting for %s to get external IP", serviceName)
	if _, err := retry.Retry(ctx, retryFn); err != nil {
		return "", err
	}

	ctx, cancel := context.WithTimeout(ctx, 300*time.Second)
	defer cancel()

	if !sanityCheck {
		return ingress, nil
	}

	client := &http.Client{Timeout: 5 * time.Second}
	ingressURL := fmt.Sprintf("http://%s", ingress)
	log.Infof("Sanity checking %v", ingressURL)
	for {
		select {
		case <-ctx.Done():
			return "", fmt.Errorf("%s readiness check timed out", serviceName)
		default:
			response, err := ctxhttp.Get(ctx, client, ingressURL)
			if err == nil {
				log.Infof("Response %v %q received from %v", response.StatusCode, response.Status, ingressURL)
				return ingress, nil
			}
		}
	}
}

func getServiceLoadBalancer(name, namespace, kubeconfig string) (string, error) {
	ip, err := ShellSilent(
		"kubectl get svc %s -n %s -o jsonpath='{.status.loadBalancer.ingress[*].ip}' --kubeconfig=%s",
		name, namespace, kubeconfig)

	if err != nil {
		return "", err
	}

	if ip == "" {
		// This block is used for docker-desktop kubernetes
		ip, err = ShellSilent(
			"kubectl get svc %s -n %s -o jsonpath='{.status.loadBalancer.ingress[*].hostname}' --kubeconfig=%s",
			name, namespace, kubeconfig)

		if err != nil {
			return "", err
		}
		if ip == "localhost" {
			ip = "127.0.0.1"
		}
	}

	ip = strings.Trim(ip, "'")
	addr := net.ParseIP(ip)
	if addr == nil {
		return "", errors.New("ingress ip not available yet")
	}

	return ip, nil
}

func getServiceNodePort(serviceName, podLabel, namespace, kubeconfig string) (string, error) {
	ip, err := Shell(
		"kubectl get po -l istio=%s -n %s -o jsonpath='{.items[0].status.hostIP}' --kubeconfig=%s",
		podLabel, namespace, kubeconfig)

	if err != nil {
		return "", err
	}

	ip = strings.Trim(ip, "'")
	addr := net.ParseIP(ip)
	if addr == nil {
		return "", fmt.Errorf("the ip of %s is not available yet", serviceName)
	}

	port, err := getServicePort(serviceName, namespace, kubeconfig)
	if err != nil {
		return "", err
	}

	if addr.To4() == nil {
		return "[" + ip + "]" + ":" + port, nil
	}
	return ip + ":" + port, nil
}

func getServicePort(serviceName, namespace, kubeconfig string) (string, error) {
	port, err := Shell(
		"kubectl get svc %s -n %s -o jsonpath='{.spec.ports[?(@.port==%s)].nodePort}' --kubeconfig=%s",
		serviceName, namespace, ingressHTTPServicePort, kubeconfig)

	if err != nil {
		return "", err
	}

	port = strings.Trim(port, "'")
	rp := regexp.MustCompile(`^[0-9]{1,5}$`)
	if rp.FindString(port) == "" {
		err = fmt.Errorf("unable to find the port of %s", serviceName)
		log.Warna(err)
		return "", err
	}
	return port, nil
}

// GetIngressPodNames get the pod names for the Istio ingress deployment.
func GetIngressPodNames(n string, kubeconfig string) ([]string, error) {
	res, err := Shell(
		"kubectl get pod -l istio=ingress -n %s -o jsonpath='{.items[*].metadata.name}' --kubeconfig=%s",
		n, kubeconfig)
	if err != nil {
		return nil, err
	}
	res = strings.Trim(res, "'")
	return strings.Split(res, " "), nil
}

// GetPodsInfo get pods info.
func GetPodsInfo(n string, kubeconfig string, label string) (pods []PodInfo, err error) {
	res, err := Shell("kubectl -n %s -l=%s get pods -o=jsonpath='{range .items[*]}{.metadata.name}{\" \"}"+
		"{.status.podIP}{\"\\n\"}{end}' --kubeconfig=%s", n, label, kubeconfig)
	if err != nil {
		log.Infof("Failed to get pods by label %s in namespace %s: %s", label, n, err)
		return nil, err
	}
	for _, line := range strings.Split(res, "\n") {
		f := strings.Fields(line)
		if len(f) >= 2 {
			pods = append(pods, PodInfo{
				Name:   f[0],
				IPAddr: f[1],
			})
		}
	}
	return pods, nil
}

// GetAppPodsInfo returns a map of a list of PodInfo
func GetAppPodsInfo(n string, kubeconfig string, label string) ([]string, map[string][]string, error) {
	// This will return a table where c0=pod_name and c1=label_value and c2=IPAddr.
	// The columns are separated by a space and each result is on a separate line (separated by '\n').
	res, err := Shell("kubectl -n %s -l=%s get pods -o=jsonpath='{range .items[*]}{.metadata.name}{\" \"}{"+
		".metadata.labels.%s}{\" \"}{.status.podIP}{\"\\n\"}{end}' --kubeconfig=%s", n, label, label, kubeconfig)
	if err != nil {
		log.Infof("Failed to get pods by label %s in namespace %s: %s", label, n, err)
		return nil, nil, err
	}

	var podNames []string
	eps := make(map[string][]string)
	for _, line := range strings.Split(res, "\n") {
		f := strings.Fields(line)
		if len(f) >= 3 {
			podNames = append(podNames, f[0])
			eps[f[1]] = append(eps[f[1]], f[2])
		}
	}

	return podNames, eps, nil
}

// GetAppPods gets a map of app names to the pods for the app, for the given namespace
func GetAppPods(n string, kubeconfig string) (map[string][]string, error) {
	podLabels, err := GetPodLabelValues(n, "app", kubeconfig)
	if err != nil {
		return nil, err
	}

	m := make(map[string][]string)
	for podName, app := range podLabels {
		m[app] = append(m[app], podName)
	}
	return m, nil
}

// IsJobSucceeded checks whether a job for the given namespace succeeded
func IsJobSucceeded(n, name string, kubeconfig string) (bool, error) {
	succeed, err := Shell("kubectl -n %s get job  %s -o jsonpath='{.status.succeeded}' --kubeconfig=%s",
		n, name, kubeconfig)
	if err != nil {
		log.Warnf("could not get %s job: %v", name, err)
		return false, err
	}

	if len(succeed) != 0 {
		return true, nil
	}
	return false, nil
}

// GetPodLabelValues gets a map of pod name to label value for the given label and namespace
func GetPodLabelValues(n, label string, kubeconfig string) (map[string]string, error) {
	// This will return a table where c0=pod_name and c1=label_value.
	// The columns are separated by a space and each result is on a separate line (separated by '\n').
	res, err := Shell("kubectl -n %s -l=%s get pods -o=jsonpath='{range .items[*]}{.metadata.name}{\" \"}{"+
		".metadata.labels.%s}{\"\\n\"}{end}' --kubeconfig=%s", n, label, label, kubeconfig)
	if err != nil {
		log.Infof("Failed to get pods by label %s in namespace %s: %s", label, n, err)
		return nil, err
	}

	// Split the lines in the result
	m := make(map[string]string)
	for _, line := range strings.Split(res, "\n") {
		f := strings.Fields(line)
		if len(f) >= 2 {
			m[f[0]] = f[1]
		}
	}

	return m, nil
}

// GetPodNames gets names of all pods in specific namespace and return in a slice
func GetPodNames(n string) (pods []string, kubeconfig string) {
	res, err := Shell("kubectl -n %s get pods -o jsonpath='{.items[*].metadata.name}' --kubeconfig=%s",
		n, kubeconfig)
	if err != nil {
		log.Infof("Failed to get pods name in namespace %s: %s", n, err)
		return
	}
	res = strings.Trim(res, "'")
	pods = strings.Split(res, " ")
	log.Infof("Existing pods: %v", pods)
	return
}

// GetPodStatus gets status of a pod from a namespace
// Note: It is not enough to check pod phase, which only implies there is at
// least one container running. Use kubectl CLI to get status so that we can
// ensure that all containers are running.
func GetPodStatus(n, pod string, kubeconfig string) string {
	status, err := Shell("kubectl -n %s get pods %s --no-headers --kubeconfig=%s", n, pod, kubeconfig)
	if err != nil {
		log.Infof("Failed to get status of pod %s in namespace %s: %s", pod, n, err)
		status = podFailedGet
	}
	f := strings.Fields(status)
	if len(f) > statusField {
		return f[statusField]
	}
	return ""
}

// GetPodName gets the pod name for the given namespace and label selector
func GetPodName(n, labelSelector string, kubeconfig string) (pod string, err error) {
	pod, err = Shell("kubectl -n %s get pod -l %s -o jsonpath='{.items[0].metadata.name}' --kubeconfig=%s",
		n, labelSelector, kubeconfig)
	if err != nil {
		log.Warnf("could not get %s pod: %v", labelSelector, err)
		return
	}
	pod = strings.Trim(pod, "'")
	log.Infof("%s pod name: %s", labelSelector, pod)
	return
}

// GetPodLogsForLabel gets the logs for the given label selector and container
func GetPodLogsForLabel(n, labelSelector string, container string, tail, alsoShowPreviousPodLogs bool,
	kubeconfig string) string {
	pod, err := GetPodName(n, labelSelector, kubeconfig)
	if err != nil {
		return ""
	}
	return GetPodLogs(n, pod, container, tail, alsoShowPreviousPodLogs, kubeconfig)
}

// GetPodLogs retrieves the logs for the given namespace, pod and container.
func GetPodLogs(n, pod, container string, tail, alsoShowPreviousPodLogs bool, kubeconfig string) string {
	tailOption := ""
	if tail {
		tailOption = "--tail=200"
	}
	o1 := ""
	if alsoShowPreviousPodLogs {
		log.Info("Expect and ignore an error getting crash logs when there are no crash (-p invocation)")
		// Do not use Shell. It dumps the entire log on the console and makes the test unusable due to very large
		// amount of output
		o1, _ = ShellMuteOutput("kubectl --namespace %s logs %s -c %s %s -p --kubeconfig=%s", n, pod, container,
			tailOption, kubeconfig)
		o1 += "\n"
	}
	// Do not use Shell. It dumps the entire log on the console and makes the test unusable due to very large amount of
	// output
	o2, _ := ShellMuteOutput("kubectl --namespace %s logs %s -c %s %s --kubeconfig=%s", n, pod, container,
		tailOption, kubeconfig)
	return o1 + o2
}

// GetConfigs retrieves the configurations for the list of resources.
func GetConfigs(kubeconfig string, names ...string) (string, error) {
	cmd := fmt.Sprintf("kubectl get %s --all-namespaces -o yaml --kubeconfig=%s", strings.Join(names, ","), kubeconfig)
	return Shell(cmd)
}

// PodExec runs the specified command on the container for the specified namespace and pod
func PodExec(n, pod, container, command string, muteOutput bool, kubeconfig string) (string, error) {
	if muteOutput {
		return ShellSilent("kubectl exec --kubeconfig=%s %s -n %s -c %s -- %s", kubeconfig, pod, n, container,
			command)
	}
	return Shell("kubectl exec --kubeconfig=%s %s -n %s -c %s -- %s ", kubeconfig, pod, n, container, command)
}

// CreateTLSSecret creates a secret from the provided cert and key files
func CreateTLSSecret(secretName, n, keyFile, certFile string, kubeconfig string) (string, error) {
	return Shell("kubectl create secret tls %s -n %s --key %s --cert %s --kubeconfig=%s", secretName, n, keyFile,
		certFile, kubeconfig)
}

// CheckPodsRunningWithMaxDuration returns if all pods in a namespace are in "Running" status
// Also check container status to be running.
func CheckPodsRunningWithMaxDuration(n string, maxDuration time.Duration, kubeconfig string) (ready bool) {
	if err := WaitForDeploymentsReady(n, maxDuration, kubeconfig); err != nil {
		log.Errorf("CheckPodsRunning: %v", err.Error())
		return false
	}

	return true
}

// CheckPodsRunning returns readiness of all pods within a namespace. It will wait for upto 2 mins.
// use WithMaxDuration to specify a duration.
func CheckPodsRunning(n string, kubeconfig string) (ready bool) {
	return CheckPodsRunningWithMaxDuration(n, 2*time.Minute, kubeconfig)
}

// CheckDeployment gets status of a deployment from a namespace
func CheckDeployment(ctx context.Context, namespace, deployment string, kubeconfig string) error {
	if deployment == "deployments/istio-sidecar-injector" {
		// This can be deployed by previous tests, but doesn't complete currently, blocking the test.
		return nil
	}
	errc := make(chan error, 1)
	go func() {
		if _, err := ShellMuteOutput("kubectl -n %s rollout status %s --kubeconfig=%s", namespace, deployment,
			kubeconfig); err != nil {
			errc <- fmt.Errorf("%s in namespace %s failed", deployment, namespace)
		}
		errc <- nil
	}()
	select {
	case err := <-errc:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// CheckAppDeployment checks whether or not an app in a namespace is ready
func CheckAppDeployment(namespace, deployment string, timeout time.Duration, kubeconfig string) error {
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return CheckDeployment(ctx, namespace, "deployments/"+deployment, kubeconfig) })
	return g.Wait()
}

// CheckDeploymentRemoved waits until a deployment is removed or times out
func CheckDeploymentRemoved(namespace, deployment string, kubeconfig string) error {
	retry := Retrier{
		BaseDelay: 5 * time.Second,
		MaxDelay:  5 * time.Second,
		Retries:   60,
	}

	pod, err := GetPodName(namespace, "name="+deployment, kubeconfig)
	// Pod has been removed
	if err != nil {
		log.Infof("pod %s is successfully removed", pod)
		return nil
	}
	retryFn := func(_ context.Context, i int) error {
		_, err := Shell("kubectl get pods %s -n %s --kubeconfig=%s", pod, namespace, kubeconfig)
		if err != nil {
			log.Infof("pod %s is successfully removed", pod)
			return nil
		}
		return fmt.Errorf("%s in namespace %s still exists", pod, namespace)
	}
	ctx := context.Background()
	_, err = retry.Retry(ctx, retryFn)
	if err != nil {
		return err
	}
	return nil
}

// CheckDeployments checks whether all deployments in a given namespace are ready
func CheckDeployments(namespace string, timeout time.Duration, kubeconfig string) error {
	// wait for istio-system deployments to be fully rolled out before proceeding
	out, err := Shell("kubectl -n %s get deployment -o name --kubeconfig=%s", namespace, kubeconfig)
	if err != nil {
		return fmt.Errorf("could not list deployments in namespace %q", namespace)
	}
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	g, ctx := errgroup.WithContext(ctx)
	deployments := strings.Fields(out)
	deploymentStart := time.Now()
	for i := range deployments {
		deployment := deployments[i]
		g.Go(func() error { return CheckDeployment(ctx, namespace, deployment, kubeconfig) })
	}
	gerr := g.Wait()
	t := time.Now()
	elapsed := t.Sub(deploymentStart)
	log.Infof("Deployment rollout ends after [%v] with err [%v]", elapsed, gerr)
	return gerr
}

// WaitForDeploymentsReady wait up to 'timeout' duration
// return an error if deployments are not ready
func WaitForDeploymentsReady(ns string, timeout time.Duration, kubeconfig string) error {
	retry := Retrier{
		BaseDelay:   10 * time.Second,
		MaxDelay:    10 * time.Second,
		MaxDuration: timeout,
		Retries:     20,
	}

	_, err := retry.Retry(context.Background(), func(_ context.Context, _ int) error {
		nr, err := CheckDeploymentsReady(ns, kubeconfig)
		if err != nil {
			return &Break{err}
		}

		if nr == 0 { // done
			return nil
		}
		return fmt.Errorf("%d deployments not ready", nr)
	})
	return err
}

// WaitForStatefulsetReady wait up to 'timeout' duration
// return an error if statefulsets are not ready
func WaitForStatefulsetReady(ns string, timeout time.Duration, kubeconfig string) error {
	retry := Retrier{
		BaseDelay:   10 * time.Second,
		MaxDelay:    10 * time.Second,
		MaxDuration: timeout,
		Retries:     20,
	}

	_, err := retry.Retry(context.Background(), func(_ context.Context, _ int) error {
		nr, err := CheckStatefulsetReady(ns, kubeconfig)
		if err != nil {
			return &Break{err}
		}

		if nr == 0 { // done
			return nil
		}
		return fmt.Errorf("%d statefulset not ready", nr)
	})
	return err
}

// CheckDeploymentsReady checks if deployment resources are ready.
// get podsReady() sometimes gets pods created by the "Job" resource which never reach the "Running" steady state.
func CheckDeploymentsReady(ns string, kubeconfig string) (int, error) {
	CMD := "kubectl -n %s get deployments -o jsonpath='{range .items[*]}{@.metadata.name}{\" \"}" +
		"{@.status.availableReplicas}{\"\\n\"}{end}' --kubeconfig=%s"
	out, err := Shell(fmt.Sprintf(CMD, ns, kubeconfig))

	if err != nil {
		return 0, fmt.Errorf("could not list deployments in namespace %q: %v", ns, err)
	}

	notReady := 0
	for _, line := range strings.Split(out, "\n") {
		flds := strings.Fields(line)
		if len(flds) < 1 {
			continue
		}
		if len(flds) > 2 {
			notReady++
			continue
		}
		if len(flds) == 1 || flds[1] == "0" { // no replicas ready
			notReady++
			continue
		}
		if _, err = strconv.Atoi(flds[1]); err != nil {
			notReady++
			continue
		}
	}

	if notReady == 0 {
		log.Infof("All deployments are ready")
	}
	return notReady, nil
}

// CheckStatefulsetReady checks if statefulset resources are ready.
// get podsReady() sometimes gets pods created by the "Job" resource which never reach the "Running" steady state.
func CheckStatefulsetReady(ns string, kubeconfig string) (int, error) {
	CMD := "kubectl -n %s get statefulset -o jsonpath='{range .items[*]}{@.metadata.name}{\" \"}" +
		"{@.status.replicas}{\" \"}{@.status.readyReplicas}{\"\\n\"}{end}' --kubeconfig=%s"
	out, err := Shell(fmt.Sprintf(CMD, ns, kubeconfig))

	if err != nil {
		return 0, fmt.Errorf("could not list statefulsets in namespace %q: %v", ns, err)
	}

	notReady := 0
	for _, line := range strings.Split(out, "\n") {
		flds := strings.Fields(line)
		if len(flds) < 1 {
			continue
		}
		if len(flds) <= 2 { // no replicas ready
			notReady++
		}
		if len(flds) == 3 {
			if flds[1] != flds[2] {
				notReady++
			}
		}
	}

	if notReady == 0 {
		log.Infof("All statefulsets are ready")
	}
	return notReady, nil
}

// GetKubeConfig will create a kubeconfig file based on the active environment the test is run in
func GetKubeConfig(filename string) error {
	_, err := ShellMuteOutput("kubectl config view --raw=true --minify=true > %s", filename)
	if err != nil {
		return err
	}
	log.Infof("kubeconfig file %s created\n", filename)
	return nil
}

// CheckPodRunning return if a given pod with labeled name in a namespace are in "Running" status
func CheckPodRunning(n, name string, kubeconfig string) error {
	retry := Retrier{
		BaseDelay: 30 * time.Second,
		MaxDelay:  30 * time.Second,
		Retries:   6,
	}

	retryFn := func(_ context.Context, i int) error {
		pod, err := GetPodName(n, name, kubeconfig)
		if err != nil {
			return err
		}
		ready := true
		if status := GetPodStatus(n, pod, kubeconfig); status != "Running" {
			log.Infof("%s in namespace %s is not running: %s", pod, n, status)
			ready = false
		}

		if !ready {
			return fmt.Errorf("pod %s is not ready", pod)
		}
		return nil
	}
	ctx := context.Background()
	_, err := retry.Retry(ctx, retryFn)
	if err != nil {
		return err
	}
	log.Infof("Got the pod name=%s running!", name)
	return nil
}
