monitor velero logs and fix E2E issues

1. Capture Velero pod log and K8S cluster event;
2. Fix wrong path of storageclass yaml file issue caused by pert test;
3. Fix change storageclass test issue that no sc named 'default' in EKS cluster;
4. Support AWS credential as config format;
5. Support more E2E script input parameters like standy cluster plugins and provider.

Signed-off-by: danfengl <danfengl@vmware.com>

test/util/common/common.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"os"
 	"os/exec"
 )
 
@@ -87,3 +88,25 @@ func CMDExecWithOutput(checkCMD *exec.Cmd) (*[]byte, error) {
 	}
 	return &jsonBuf, err
 }
+
+func WriteToFile(content, fileName string) error {
+	file, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE, 0666)
+	if err != nil {
+		fmt.Println("fail to open file", err)
+		return err
+	}
+	defer file.Close()
+
+	write := bufio.NewWriter(file)
+	_, err = write.WriteString(content)
+	if err != nil {
+		fmt.Println("fail to WriteString file", err)
+		return err
+	}
+	err = write.Flush()
+	if err != nil {
+		fmt.Println("fail to Flush file", err)
+		return err
+	}
+	return nil
+}

test/util/k8s/common.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"strings"
 	"time"
 
 	"github.com/pkg/errors"
@@ -30,7 +31,7 @@ import (
 
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	veleroexec "github.com/vmware-tanzu/velero/pkg/util/exec"
-	"github.com/vmware-tanzu/velero/test/util/common"
+	common "github.com/vmware-tanzu/velero/test/util/common"
 )
 
 // ensureClusterExists returns whether or not a Kubernetes cluster exists for tests to be run on.
@@ -289,15 +290,15 @@ func ReadFileFromPodVolume(ctx context.Context, namespace, podName, containerNam
 	return stdout, err
 }
 
-func KubectlGetInfo(cmdName string, arg []string) {
+func RunCommand(cmdName string, arg []string) string {
 	cmd := exec.CommandContext(context.Background(), cmdName, arg...)
-	fmt.Printf("Kubectl exec cmd =%v\n", cmd)
+	fmt.Printf("Run cmd =%v\n", cmd)
 	stdout, stderr, err := veleroexec.RunCommand(cmd)
-	fmt.Println(stdout)
 	if err != nil {
 		fmt.Println(stderr)
 		fmt.Println(err)
 	}
+	return stdout
 }
 
 func KubectlGetDsJson(veleroNamespace string) (string, error) {
@@ -363,3 +364,26 @@ func CreateVolumes(pvcName string, volumeNameList []string) (vols []*corev1.Volu
 	}
 	return
 }
+
+func CollectClusterEvents(key string, pods []string) {
+	prefix := "pod"
+	date := RunCommand("date", []string{"-u"})
+	logs := []string{}
+	logs = append(logs, date)
+	logs = append(logs, RunCommand("kubectl", []string{"get", "events", "-o", "custom-columns=FirstSeen:.firstTimestamp,Count:.count,From:.source.component,Type:.type,Reason:.reason,Message:.message", "--all-namespaces"}))
+	logs = append(logs, RunCommand("kubectl", []string{"get", "events", "-o", "yaml", "--all-namespaces"}))
+	for _, pod := range pods {
+		logs = append(logs, RunCommand("kubectl", []string{"logs", "-n", "velero", pod, "--previous"}))
+		prefix = fmt.Sprintf("%s-%s", prefix, pod)
+	}
+	logs = append(logs, RunCommand("date", []string{"-u"}))
+	log := strings.Join(logs, "\n")
+
+	fileName := fmt.Sprintf("%s-%s", prefix, key)
+	fmt.Printf("Cluster event log file %s: %s", fileName, log)
+
+	err := common.WriteToFile(log, fileName)
+	if err != nil {
+		fmt.Printf("Fail to log cluster event: %v", err)
+	}
+}

test/util/kibishii/kibishii_utils.go

@@ -105,6 +105,7 @@ func RunKibishiiTests(veleroCfg VeleroConfig, backupName, restoreName, backupLoc
 		RunDebug(context.Background(), veleroCLI, veleroNamespace, backupName, "")
 		return errors.Wrapf(err, "Failed to backup kibishii namespace %s", kibishiiNamespace)
 	}
+
 	fmt.Printf("VeleroBackupNamespace done %s\n", time.Now().Format("2006-01-02 15:04:05"))
 	if useVolumeSnapshots {
 		if providerName == "vsphere" {
@@ -291,15 +292,16 @@ func verifyData(ctx context.Context, namespace string, kibishiiData *KibishiiDat
 			return false, nil
 		}
 		if err != nil {
-			fmt.Printf("Kibishi verify stdout Timeout occurred: %s stderr: %s err: %s", stdout, stderr, err)
+			fmt.Printf("Kibishi verify stdout Timeout occurred: %s stderr: %s err: %s\n", stdout, stderr, err)
 			return false, nil
 		}
 		return true, nil
 	})
 
 	if err != nil {
-		return errors.Wrapf(err, fmt.Sprintf("Failed to wait generate data in namespace %s", namespace))
+		return errors.Wrapf(err, fmt.Sprintf("Failed to verify kibishii data in namespace %s\n", namespace))
 	}
+	fmt.Printf("Success to verify kibishii data in namespace %s\n", namespace)
 	return nil
 }
 

test/util/providers/aws_utils.go

@@ -18,15 +18,20 @@ package providers
 
 import (
 	"fmt"
+	"net/url"
+	"os"
+	"strconv"
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/endpoints"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 	"github.com/pkg/errors"
+	"golang.org/x/net/context"
 
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
 	"github.com/vmware-tanzu/velero/test"
@@ -34,6 +39,173 @@ import (
 
 type AWSStorage string
 
+const (
+	s3URLKey                     = "s3Url"
+	publicURLKey                 = "publicUrl"
+	kmsKeyIDKey                  = "kmsKeyId"
+	customerKeyEncryptionFileKey = "customerKeyEncryptionFile"
+	s3ForcePathStyleKey          = "s3ForcePathStyle"
+	bucketKey                    = "bucket"
+	signatureVersionKey          = "signatureVersion"
+	credentialsFileKey           = "credentialsFile"
+	credentialProfileKey         = "profile"
+	serverSideEncryptionKey      = "serverSideEncryption"
+	insecureSkipTLSVerifyKey     = "insecureSkipTLSVerify"
+	caCertKey                    = "caCert"
+	enableSharedConfigKey        = "enableSharedConfig"
+)
+
+func newSessionOptions(config aws.Config, profile, caCert, credentialsFile, enableSharedConfig string) (session.Options, error) {
+	sessionOptions := session.Options{Config: config, Profile: profile}
+	if caCert != "" {
+		sessionOptions.CustomCABundle = strings.NewReader(caCert)
+	}
+
+	if credentialsFile == "" && os.Getenv("AWS_SHARED_CREDENTIALS_FILE") != "" {
+		credentialsFile = os.Getenv("AWS_SHARED_CREDENTIALS_FILE")
+	}
+
+	if credentialsFile != "" {
+		if _, err := os.Stat(credentialsFile); err != nil {
+			if os.IsNotExist(err) {
+				return session.Options{}, errors.Wrapf(err, "provided credentialsFile does not exist")
+			}
+			return session.Options{}, errors.Wrapf(err, "could not get credentialsFile info")
+		}
+		sessionOptions.SharedConfigFiles = append(sessionOptions.SharedConfigFiles, credentialsFile)
+		sessionOptions.SharedConfigState = session.SharedConfigEnable
+	}
+
+	return sessionOptions, nil
+}
+
+// takes AWS session options to create a new session
+func getSession(options session.Options) (*session.Session, error) {
+	sess, err := session.NewSessionWithOptions(options)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	if _, err := sess.Config.Credentials.Get(); err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return sess, nil
+}
+
+// GetBucketRegion returns the AWS region that a bucket is in, or an error
+// if the region cannot be determined.
+func GetBucketRegion(bucket string) (string, error) {
+	var region string
+
+	session, err := session.NewSession()
+	if err != nil {
+		return "", errors.WithStack(err)
+	}
+
+	for _, partition := range endpoints.DefaultPartitions() {
+		for regionHint := range partition.Regions() {
+			region, _ = s3manager.GetBucketRegion(context.Background(), session, bucket, regionHint)
+
+			// we only need to try a single region hint per partition, so break after the first
+			break
+		}
+
+		if region != "" {
+			return region, nil
+		}
+	}
+
+	return "", errors.New("unable to determine bucket's region")
+}
+
+func (s AWSStorage) CreateSession(credentialProfile, credentialsFile, enableSharedConfig, caCert, bucket, bslPrefix, bslConfig string) (*session.Session, error) {
+	var err error
+	config := flag.NewMap()
+	config.Set(bslConfig)
+	region := config.Data()["region"]
+	objectsInput := s3.ListObjectsV2Input{}
+	objectsInput.Bucket = aws.String(bucket)
+	objectsInput.Delimiter = aws.String("/")
+	s3URL := ""
+	s3ForcePathStyleVal := ""
+	s3ForcePathStyle := false
+
+	if s3ForcePathStyleVal != "" {
+		if s3ForcePathStyle, err = strconv.ParseBool(s3ForcePathStyleVal); err != nil {
+			return nil, errors.Wrapf(err, "could not parse %s (expected bool)", s3ForcePathStyleKey)
+		}
+	}
+
+	// AWS (not an alternate S3-compatible API) and region not
+	// explicitly specified: determine the bucket's region
+	if s3URL == "" && region == "" {
+		var err error
+
+		region, err = GetBucketRegion(bucket)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	serverConfig, err := newAWSConfig(s3URL, region, s3ForcePathStyle)
+	if err != nil {
+		return nil, err
+	}
+
+	sessionOptions, err := newSessionOptions(*serverConfig, credentialProfile, caCert, credentialsFile, enableSharedConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	serverSession, err := getSession(sessionOptions)
+
+	if err != nil {
+		fmt.Println(err)
+		return nil, err
+	}
+
+	return serverSession, nil
+}
+
+// IsValidS3URLScheme returns true if the scheme is http:// or https://
+// and the url parses correctly, otherwise, return false
+func IsValidS3URLScheme(s3URL string) bool {
+	u, err := url.Parse(s3URL)
+	if err != nil {
+		return false
+	}
+	if u.Scheme != "http" && u.Scheme != "https" {
+		return false
+	}
+	return true
+}
+func newAWSConfig(url, region string, forcePathStyle bool) (*aws.Config, error) {
+	awsConfig := aws.NewConfig().
+		WithRegion(region).
+		WithS3ForcePathStyle(forcePathStyle)
+
+	if url != "" {
+		if !IsValidS3URLScheme(url) {
+			return nil, errors.Errorf("Invalid s3 url %s, URL must be valid according to https://golang.org/pkg/net/url/#Parse and start with http:// or https://", url)
+		}
+
+		awsConfig = awsConfig.WithEndpointResolver(
+			endpoints.ResolverFunc(func(service, region string, optFns ...func(*endpoints.Options)) (endpoints.ResolvedEndpoint, error) {
+				if service == s3.EndpointsID {
+					return endpoints.ResolvedEndpoint{
+						URL: url,
+					}, nil
+				}
+
+				return endpoints.DefaultResolver().EndpointFor(service, region, optFns...)
+			}),
+		)
+	}
+
+	return awsConfig, nil
+}
+
 func (s AWSStorage) ListItems(client *s3.S3, objectsV2Input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
 	res, err := client.ListObjectsV2(objectsV2Input)
 	if err != nil {
@@ -54,15 +226,20 @@ func (s AWSStorage) DeleteItem(client *s3.S3, deleteObjectV2Input *s3.DeleteObje
 func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) (bool, error) {
 	config := flag.NewMap()
 	config.Set(bslConfig)
-	region := config.Data()["region"]
 	objectsInput := s3.ListObjectsV2Input{}
 	objectsInput.Bucket = aws.String(bslBucket)
 	objectsInput.Delimiter = aws.String("/")
-	s3url := ""
+
 	if bslPrefix != "" {
 		objectsInput.Prefix = aws.String(bslPrefix)
 	}
+
+	var err error
 	var s3Config *aws.Config
+	var sess *session.Session
+	region := config.Data()["region"]
+	s3url := ""
+
 	if region == "minio" {
 		s3url = config.Data()["s3Url"]
 		s3Config = &aws.Config{
@@ -72,21 +249,19 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 			DisableSSL:       aws.Bool(true),
 			S3ForcePathStyle: aws.Bool(true),
 		}
+		sess, err = session.NewSession(s3Config)
 	} else {
-		s3Config = &aws.Config{
-			Region:      aws.String(region),
-			Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
-		}
+		sess, err = s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
 	}
-	sess, err := session.NewSession(s3Config)
 
 	if err != nil {
-		return false, errors.Wrapf(err, "Failed to create AWS session")
+		return false, errors.Wrapf(err, fmt.Sprintf("Failed to create AWS session of region %s", region))
 	}
 	svc := s3.New(sess)
 
 	bucketObjects, err := s.ListItems(svc, &objectsInput)
 	if err != nil {
+		fmt.Println("Couldn't retrieve bucket items!")
 		return false, errors.Wrapf(err, "Couldn't retrieve bucket items")
 	}
 
@@ -111,12 +286,17 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) error {
 	config := flag.NewMap()
 	config.Set(bslConfig)
+
+	var err error
+	var sess *session.Session
 	region := config.Data()["region"]
 	s3url := ""
+
 	s3Config := &aws.Config{
 		Region:      aws.String(region),
 		Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
 	}
+
 	if region == "minio" {
 		s3url = config.Data()["s3Url"]
 		s3Config = &aws.Config{
@@ -126,12 +306,17 @@ func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 			DisableSSL:       aws.Bool(true),
 			S3ForcePathStyle: aws.Bool(true),
 		}
+		sess, err = session.NewSession(s3Config)
+	} else {
+		sess, err = s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
 	}
-	sess, err := session.NewSession(s3Config)
+
 	if err != nil {
-		return errors.Wrapf(err, "Error waiting for uploads to complete")
+		return errors.Wrapf(err, fmt.Sprintf("Failed to create AWS session of region %s", region))
 	}
+
 	svc := s3.New(sess)
+
 	fullPrefix := strings.Trim(bslPrefix, "/") + "/" + strings.Trim(backupObject, "/") + "/"
 	iter := s3manager.NewDeleteListIterator(svc, &s3.ListObjectsInput{
 		Bucket: aws.String(bslBucket),
@@ -139,7 +324,7 @@ func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 	})
 
 	if err := s3manager.NewBatchDeleteWithClient(svc).Delete(aws.BackgroundContext(), iter); err != nil {
-		return errors.Wrapf(err, "Error waiting for uploads to complete")
+		return errors.Wrapf(err, "Fail to delete object")
 	}
 	fmt.Printf("Deleted object(s) from bucket: %s %s \n", bslBucket, fullPrefix)
 	return nil
@@ -150,16 +335,15 @@ func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObj
 	config := flag.NewMap()
 	config.Set(bslConfig)
 	region := config.Data()["region"]
-	s3Config := &aws.Config{
-		Region:      aws.String(region),
-		Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
-	}
+
 	if region == "minio" {
 		return errors.New("No snapshot for Minio provider")
 	}
-	sess, err := session.NewSession(s3Config)
+	sess, err := s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
+
 	if err != nil {
-		return errors.Wrapf(err, "Error waiting for uploads to complete")
+		fmt.Printf("Fail to create session with profile %s and config %s", cloudCredentialsFile, bslConfig)
+		return errors.Wrapf(err, "Fail to create session with profile %s and config %s", cloudCredentialsFile, bslConfig)
 	}
 	svc := ec2.New(sess)
 	params := &ec2.DescribeSnapshotsInput{

test/util/velero/install.go

@@ -240,7 +240,7 @@ func installVeleroServer(ctx context.Context, cli, cloudProvider string, options
 		args = append(args, "--features", options.Features)
 		if strings.EqualFold(options.Features, "EnableCSI") && options.UseVolumeSnapshots {
 			if strings.EqualFold(cloudProvider, "azure") {
-				if err := KubectlApplyByFile(ctx, "util/csi/AzureVolumeSnapshotClass.yaml"); err != nil {
+				if err := KubectlApplyByFile(ctx, "../util/csi/AzureVolumeSnapshotClass.yaml"); err != nil {
 					return err
 				}
 			}

test/util/velero/velero_utils.go

@@ -217,8 +217,14 @@ func checkBackupPhase(ctx context.Context, veleroCLI string, veleroNamespace str
 		return err
 	}
 	if backup.Status.Phase != expectedPhase {
+		if VeleroCfg.DebugVeleroPodRestart {
+			pods, err := GetVeleroPodName(ctx)
+			fmt.Println(err)
+			CollectClusterEvents(backupName, pods)
+		}
 		return errors.Errorf("Unexpected backup phase got %s, expecting %s", backup.Status.Phase, expectedPhase)
 	}
+
 	return nil
 }
 
@@ -239,6 +245,11 @@ func checkRestorePhase(ctx context.Context, veleroCLI string, veleroNamespace st
 		return err
 	}
 	if restore.Status.Phase != expectedPhase {
+		if VeleroCfg.DebugVeleroPodRestart {
+			pods, err := GetVeleroPodName(ctx)
+			fmt.Println(err)
+			CollectClusterEvents(restoreName, pods)
+		}
 		return errors.Errorf("Unexpected restore phase got %s, expecting %s", restore.Status.Phase, expectedPhase)
 	}
 	return nil
@@ -442,7 +453,6 @@ func VeleroRestoreExec(ctx context.Context, veleroCLI, veleroNamespace, restoreN
 	if err := VeleroCmdExec(ctx, veleroCLI, args); err != nil {
 		return err
 	}
-
 	return checkRestorePhase(ctx, veleroCLI, veleroNamespace, restoreName, phaseExpect)
 }
 
@@ -1241,7 +1251,7 @@ func GetVersionList(veleroCli, veleroVersion string) []VeleroCLI2Version {
 	}
 	return veleroCLI2VersionList
 }
-func DeleteBackups(ctx context.Context, client TestClient) error {
+func DeleteAllBackups(ctx context.Context, client TestClient) error {
 	backupList := new(velerov1api.BackupList)
 	if err := client.Kubebuilder.List(ctx, backupList, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
 		return fmt.Errorf("failed to list backup object in %s namespace with err %v", VeleroCfg.VeleroNamespace, err)
@@ -1255,6 +1265,24 @@ func DeleteBackups(ctx context.Context, client TestClient) error {
 	return nil
 }
 
+func DeleteBackups(ctx context.Context, client TestClient, backupNames []string) error {
+	backupList := new(velerov1api.BackupList)
+	if err := client.Kubebuilder.List(ctx, backupList, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
+		return fmt.Errorf("failed to list backup object in %s namespace with err %v", VeleroCfg.VeleroNamespace, err)
+	}
+	for _, backup := range backupList.Items {
+		for _, bn := range backupNames {
+			if backup.Name == bn {
+				fmt.Printf("Backup %s is going to be deleted...\n", backup.Name)
+				if err := VeleroBackupDelete(ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, backup.Name); err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
 func DeleteRestores(ctx context.Context, client TestClient) error {
 	restoreList := new(velerov1api.RestoreList)
 	if err := client.Kubebuilder.List(ctx, restoreList, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
@@ -1482,3 +1510,56 @@ func IsSupportUploaderType(version string) (bool, error) {
 		return false, nil
 	}
 }
+
+func GetVeleroPodName(ctx context.Context) ([]string, error) {
+	// Example:
+	//    NAME                                  STATUS   VOLUME                                     CAPACITY   ACCESS MODES   STORAGECLASS             AGE
+	//    kibishii-data-kibishii-deployment-0   Bound    pvc-94b9fdf2-c30f-4a7b-87bf-06eadca0d5b6   1Gi        RWO            kibishii-storage-class   115s
+	cmds := []*common.OsCommandLine{}
+	cmd := &common.OsCommandLine{
+		Cmd:  "kubectl",
+		Args: []string{"get", "pod", "-n", "velero"},
+	}
+	cmds = append(cmds, cmd)
+
+	cmd = &common.OsCommandLine{
+		Cmd:  "grep",
+		Args: []string{"velero"},
+	}
+	cmds = append(cmds, cmd)
+
+	cmd = &common.OsCommandLine{
+		Cmd:  "awk",
+		Args: []string{"{print $1}"},
+	}
+	cmds = append(cmds, cmd)
+
+	return common.GetListByCmdPipes(ctx, cmds)
+}
+
+func InstallTestStorageClasses(path string) error {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), time.Minute*5)
+	defer ctxCancel()
+	err := InstallStorageClass(ctx, path)
+	if err != nil {
+		return err
+	}
+	content, err := os.ReadFile(path)
+	if err != nil {
+		return errors.Wrapf(err, "failed to get %s when install storage class", path)
+	}
+
+	// replace sc to new value
+	newContent := strings.ReplaceAll(string(content), fmt.Sprintf("name: %s", StorageClassName), fmt.Sprintf("name: %s", StorageClassName2))
+
+	tmpFile, err := os.CreateTemp("", "sc-file")
+	if err != nil {
+		return errors.Wrapf(err, "failed to create temp file  when install storage class")
+	}
+
+	defer os.Remove(tmpFile.Name())
+	if _, err := tmpFile.WriteString(newContent); err != nil {
+		return errors.Wrapf(err, "failed to write content into temp file %s when install storage class", tmpFile.Name())
+	}
+	return InstallStorageClass(ctx, tmpFile.Name())
+}